package org.activequant.util.ib;

import java.io.DataInputStream;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.GregorianCalendar;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.StringTokenizer;
import java.util.TimeZone;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

import org.activequant.calendar.NASDAQCalendarSettings;
import org.activequant.core.domainmodel.InstrumentSpecification;
import org.activequant.core.domainmodel.SeriesSpecification;
import org.activequant.core.domainmodel.Symbol;
import org.activequant.core.domainmodel.data.Candle;
import org.activequant.core.domainmodel.data.CandleSeries;
import org.activequant.core.domainmodel.data.Quote;
import org.activequant.core.domainmodel.data.QuoteSeries;
import org.activequant.core.domainmodel.data.TradeIndication;
import org.activequant.core.domainmodel.marketdepth.MarketDepthEvent;
import org.activequant.core.domainmodel.marketscanner.InstrumentSpecificationDetails;
import org.activequant.core.domainmodel.marketscanner.MarketScan;
import org.activequant.core.domainmodel.marketscanner.ScanCriteria;
import org.activequant.core.types.Currency;
import org.activequant.core.types.Expiry;
import org.activequant.core.types.SecurityType;
import org.activequant.core.types.TimeFrame;
import org.activequant.core.types.TimeStamp;
import org.activequant.core.util.TimeSeriesUtils;
import org.activequant.core.util.TimeStampFormat;
import org.activequant.data.retrieval.ib.IBNewsBulletin;
import org.activequant.data.retrieval.ib.IBWrapper;
import org.activequant.util.ib.marketscanners.ScannerUtil;
import org.activequant.util.pattern.events.Event;
import org.activequant.util.pattern.events.IEventListener;
import org.activequant.util.tools.StackTraceParser;
import org.activequant.util.tools.UniqueDateGenerator;
import org.apache.log4j.Logger;

import com.ib.client.Contract;
import com.ib.client.ContractDetails;
import com.ib.client.EClientSocket;
import com.ib.client.EReader;
import com.ib.client.EWrapper;
import com.ib.client.Execution;
import com.ib.client.ExecutionFilter;
import com.ib.client.Order;
import com.ib.client.OrderState;
import com.ib.client.ScannerSubscription;
import com.ib.client.TickType;
import com.ib.client.UnderComp;

/**
 * 
 * Don't use this class directly, always wrap it as in IBCandleSeriesSource. 
 * moved here from ccapi2.  <br>
 * <br>
 * IBTwsConnection extends Thread. Holds the following associated variables:
 * <ul>
 * <li>interceptors(Map&lt;Integer,EWrapper&gt;)</li>
 * <li>connected(boolean)</li>
 * <li>autoReconnect(boolean)</li>
 * <li>reconnectInProgress(boolean)</li>
 * <li>ibEventListeners(Queue&lt;IBEventListener&gt;)</li>
 * <li>ecs(EClientSocket)</li>
 * <li>ip(String)</li>
 * <li>clientId(int)</li>
 * <li>port(int)</li>
 * <li>orderId(int)</li>
 * <li>historicRequestMemoryQueue(BlockingQueue&lt;IBHistoricalDataRequestEvent&gt;)</li>
 * <li>removeExpiredHistoricalRequests(boolean)</li>
 * <li>requestAccountDataPeriodicaly(boolean)</li>
 * <li>previouslyOpenedOrdersLoaded(boolean)</li>
 * <li>myWrapper(EWrapper)</li>
 * <li>requestId(AtomicInteger)</li>
 * <li>tickWrappers(Map&lt;InstrumentSpecification,TickWrapper&gt;)</li>
 * <li>candleWrappers(Map&lt;SeriesSpecification,CandleWrapper&gt;)</li>
 * <li>inFetch(AtomicBoolean)</li>
 * <li>isCandleFetched(AtomicBoolean)</li>
 * <li>isQuoteFetched(AtomicBoolean)</li>
 * <li>candleSeriesBag(AtomicReference&lt;CandleSeries&gt;)</li>
 * <li>quoteSeriesBag(AtomicReference&lt;QuoteSeries&gt;)</li>
 * <li>duration(int)</li>
 * <li>fetchGranularity(String)</li>
 * <li>histCandleCallback(EWrapper)</li>
 * <li>histQuoteCallback(EWrapper)</li>
 * <li>lastFetch(AtomicLong)</li>
 * <li>stopFetchingOnEmpty(boolean)</li>
 * </ul>
 * <b>History:</b><br>
 *  - [14.06.2007] Created (Ulrich Staudinger)<br>
 *  - [23.06.2007] Moved to new domain model (Erik Nijkamp)<br>
 *  - [24.06.2007] renaming and abstraction (Ulrich Staudinger)<br>
 *  - [18.07.2007] small fix for daily data (Ulrich Staudinger)<br>
 *  - [29.09.2007] cleanup + moved to new domain model (Erik Nijkamp)<br>
 *  - [28.10.2007] moved to new data subscription model (Mike Kroutikov)<br>
 *  - [19.08.2008] Adding candle stream part (Ulrich Staudinger)<br>
 *  - [12.03.2009] Misc fixes (Eyon Land)<br>
 *  - [10.06.2009] Applied patch by Kirill Jacobson, #34 (Ulrich Staudinger)<br>
 *  
 *  @author Ulrich Staudinger
 *  @author Mike Kroutikov
 *  @author Erik Nijkamp
 */
public class IBTwsConnection extends Thread {

	/**
	 * logger. 
	 */
	protected final static Logger log = Logger.getLogger(IBTwsConnection.class);

	/**
	 * private final Map&lt;Integer,EWrapper&gt; interceptors = new ConcurrentHashMap&lt;Integer,EWrapper&gt;();<br/>
	 * demultiplexes server response and sends it to the
	 * corresponding EWrapper implementation
	 */
	private final Map<Integer,EWrapper> interceptors = new ConcurrentHashMap<Integer,EWrapper>();
	
	/**
	 * private boolean connected = false;<br/>
	 * private flag indicating whether the monitor is connected or not. 
	 */
	private boolean connected = false;
	/**
	 * private boolean autoReconnect = true;
	 */
	private boolean autoReconnect = true;
	/**
	 * private boolean reconnectInProgress = false;
	 */
	private boolean reconnectInProgress = false; 
	/**
	 * public static final String TRADES = "TRADES";
	 */
	public static final String TRADES = "TRADES";
	/**
	 * public static final String MIDPOINT = "MIDPOINT";
	 */
	public static final String MIDPOINT = "MIDPOINT";
	/**
	 * public static final String BID = "BID";
	 */
	public static final String BID = "BID";
	/**
	 * public static final String ASK = "ASK";
	 */
	public static final String ASK = "ASK";
	/**
	 * public static final String ASK = "ASK";
	 */
	public static final String BID_ASK = "BID_ASK";
	/**
	 * public static final String DATA_TYPE = "type";
	 */
	public static final String DATA_TYPE = "type";
	/**
	 * private Queue&lt;IBEventListener&gt; ibEventListeners = new ConcurrentLinkedQueue&lt;IBEventListener&gt;(); 
	 */
	private Queue<IBEventListener> ibEventListeners = new ConcurrentLinkedQueue<IBEventListener>(); 
	/**
	 * private EClientSocket ecs;
	 */
	private EClientSocket ecs;
	/**
	 * private String ip;
	 */
	private String ip;

	/**
	 * private int clientId;<br/>
	 * this connection's client id.
	 */
	private int clientId;
	/**
	 * private int port;
	 */
	private int port;

	/**
	 * private int orderId = 0;<br/>
	 * the latest order id. Will be filled up automatically shortly after connection by nextValidOrderId callback.
	 */
	private int orderId = 0;
	/**
	 * private final BlockingQueue&lt;IBHistoricalDataRequestEvent&gt; historicRequestMemoryQueue = new LinkedBlockingQueue&lt;IBHistoricalDataRequestEvent&gt;(58);
	 */
	private final BlockingQueue<IBHistoricalDataRequestEvent> historicRequestMemoryQueue = new LinkedBlockingQueue<IBHistoricalDataRequestEvent>(58);
	/**
	 * private boolean removeExpiredHistoricalRequests=true;
	 */
	private boolean removeExpiredHistoricalRequests=true;
	/**
	 * private boolean requestAccountDataPeriodicaly=false;
	 */
	private boolean requestAccountDataPeriodicaly=false;
	/**
	 * private boolean previouslyOpenedOrdersLoaded=false;<br/>
	 * This shows whether we have passed the stage of loading any open orders from a previous session 
	 */
	private boolean previouslyOpenedOrdersLoaded=false;
	/**
	 * Returns true if the given code(int) is one of: 103, 104, 105, 106, 107, 109, 110, 111, 112, 113, 114, 115, 116, 117, 118, 119, 120, 121, 122, 133, 135, 201, 202, 203, 321, 329.
	 * Otherwise returns false.
	 * @param code
	 * @return
	 */
	private static boolean isOrderRelatedErrorCode(int code) {
		switch(code) {
		case 103: // duplicate order id
		case 104: // can't modify filled order
		case 105: // order being modified does not match the original order
		case 106: // can't transmit order id
		case 107: // can't transmit incomplete order
		case 109: // price is insane. order will not be submitted
		case 110: // price does not conform to min price variation for this contract
		case 111: // tif type and order type are incompatible
		case 112: // exchange not compatible with order type
		case 113: // unrecognized tif option
		case 114: // relative orders are only valid for stocks
		case 115: // relative orders for stocks can only be submitted to SMART or INet.
		case 116: // dead exchange
		case 117: // block order size must be at least 50
		case 118: // VWAP orders must be routed thru VWAP exchange
		case 119: // only VWAP orders may be placed on VWAP exchange
		case 120: // its too late to place VWAP order for today
		case 121: // incorrect BD flag for the order
		case 122: // no request tag has been found for the order
		case 133:
		case 135: 
		case 201: // order rejected
		case 202: // order canceled
		case 203: // security is not available for this account 
		case 321: case 329:
			return true;
		default:
			return false;
		}
	}
	
	/**
	 * private final EWrapper myWrapper = new EWrapper() {...}<br/>
	 * central event dispatcher: sends events to registered interceptors.
	 * handles global events and errors.
	 */
	private final EWrapper myWrapper = new EWrapper() {
		/**
		 * only logs the given arg0(ContractDetails) in the form:<br/>
		 * "bond contract details. ${arg0}"
		 */
		public void bondContractDetails(int reqId,ContractDetails arg0) {
			log.warn("bond contract details. " + arg0);
		}
		
		public void connectionClosed() {
			connected = ecs.isConnected();
			log.warn("[connectionClosed] connection between IB and TWS has been closed, can happen.");
			
			if(autoReconnect){
				log.fatal("[disconnect] Connection TraderWorkstation lost. Attempting a reconnect.");
				reconnect();
			}
			else{
				reconnectInProgress = false; 
			}
		}

		public void contractDetails(int reqId, ContractDetails contractDetails) {			
			//log.info("[contractDetails] : " + reqId+" contractDetails: "+contractDetails);
			EWrapper w = interceptors.get(reqId);
			if(w != null) {
				w.contractDetails(reqId,contractDetails);
				return;
			}
			log.warn("[contractDetails] orphan event: " + reqId);
		}
		/**
		 * logs the given time(long) in the form:<br/>
		 * "[currentTime] ${time}"
		 */
		public void currentTime(long time) {
			log.info("[currentTime] " + time);
		}
		/**
		 * logs the given e(Exception) and prints its errorStackTrace in the form:<br/>
		 * "[error] error 1: ${e}"
		 */
		public void error(Exception e) {
			log.fatal("[error] error 1: " + e);
			e.printStackTrace();
		}
		/**
		 * logs the given message(String) in this form:<br/>
		 * "[error] error 1: ${message}"
		 */
		public void error(String message) {
			log.fatal("[error] error 1: " + message);
		}
		/**
		 * <strong>1.</strong> If the given errorCode(int) is 165, it returns without doing anything.<br/>
		 * <strong>2.</strong> If the given errorCode(int) is an Order related ErrorCode, it iterates the associated ibEventListeners(Queue&lt;IBEventListener&gt;),
		 * invokes <code>listener.error(reqId, errorCode, arg2)</code> on each listener and then returns.<br/>
		 * <strong>3.</strong> Otherwise, if there is a w(EWrapper) mapped under the given reqId(int) in the interceptors(Map&lt;Integer,EWrapper&gt;), it
		 * invokes <code>w.error(reqId, errorCode, arg2)</code> on that EWrapper and returns.<br/>
		 * <strong>4.</strong> Otherwise, if the given errorCode(int) is less than 2104 and greater than 2108, it just logs it (as fatal)<br/>
		 * <strong>5.</strong> Otherwise, if the given errorCode(int) is 1102, 1101 or 504, it re-connects and re-subscribes market data subscriptions to TWS
		 */
		public void error(int reqId, int errorCode, String arg2) {
			
			if(errorCode == 165) {
				log.info("informational error message: " + arg2);
				return;
			} else if(isOrderRelatedErrorCode(errorCode)) {
				// dispatch it. 
				for(IBEventListener listener : ibEventListeners){
					listener.error(reqId, errorCode, arg2);
				}
				return;
			}
			
			EWrapper w = interceptors.get(reqId);
			if(w != null) {
				w.error(reqId, errorCode, arg2);
				return;
			}

			if(errorCode < 2104 && errorCode > 2108) {
				log.fatal("[error] error 2: " + reqId + " / " + errorCode + " / " + arg2);
			} else if (errorCode == 1102 || errorCode == 1101 || errorCode == 504) {
				log.info("[error] reconnect");
				// this is a reconnect.
				reconnect();
			} 
			
			log.error("[feedback from tws]: " + reqId + " / " + errorCode + " / " + arg2);
		}
		/**
		 * <strong>1.</strong> Logs the execution info using the given ibExec(Execution), ibContract(Contract) and orderId(int) in the form:<br/>
		 * "[execDetails]: exec_id=${ibExec.m_execId}, symbol=${ibContract.m_symbol}, shares=${ibExec.m_shares}, price=${ibExec.m_price}, time=${ibExec.m_time}"<br/>
		 * <strong>2.</strong> Notifies each of the ibEventListeners(Queue&lt;IBEventListener&gt;) that the given ibExec(Execution) occurred for the given ibContract(Contract) and orderId(int)<br/>
		 * (by invoking the <code>listener.execDetails(orderId, ibContract, ibExec)</code> method on each listener)
		 */
		public void execDetails(int orderId, com.ib.client.Contract ibContract, Execution ibExec) {
			//log.info("[execDetails]: exec_id=" + ibExec.m_execId + ", symbol=" + ibContract.m_symbol + ", shares=" + ibExec.m_shares + ", price=" + ibExec.m_price + ", time=" + ibExec.m_time);
			// dispatch it. 
			for(IBEventListener listener : ibEventListeners){
				listener.execDetails(orderId, ibContract, ibExec);
			}
		}
		/**
		 * an empty method
		 */
		public void execDetailsEnd(int reqId) {
			//log.info("[execDetailsEnd] reqId="+reqId);
			for(IBEventListener listener : ibEventListeners){
				listener.execDetailsEnd(reqId);
			}
		}
		/**
		 * <strong>1.</strong> Gets (if there is one) the w(EWrapper) mapped under the given reqId(int) in the interceptors(Map&lt;Integer,EWrapper&gt;) and invokes
		 * <code>w.historicalData(reqId, date, open, high, low, close,volume, count, WAP, hasGaps)</code> using the given reqId(int), date(String), open(double), high(double)
		 * , low(double), close(double), volume(int), count(int), WAP(double) and hasGaps(boolean)<br/>
		 * <strong>2.</strong> Otherwise if there is no such mapping, it logs the following information:<br/>
		 * "[historicalData] orphan event: ${reqId} / ${date} / ${open} / ${high} / ${low} / ${close} / ${volume} / ${count} / ${WAP}/${hasGaps}"
		 */
		public void historicalData(int reqId, String date, double open, double high, double low, double close, int volume, int count, double WAP, boolean hasGaps) {
			EWrapper w = interceptors.get(reqId);
			if(w != null) {
				w.historicalData(reqId, date, open, high, low, close,volume, count, WAP, hasGaps);
				return;
			}
			log.warn("[historicalData] orphan event: " + reqId + " / " + date + " / " + open + " / " + high + " / " + low + " / " + close + " / " + volume + " / " + count + " / " + WAP + "/" + hasGaps);
		}
		/**
		 * logs the given arg0(String) in this form:<br/>
		 * "[managedAccounts] ${arg0}"
		 */
		public void managedAccounts(String arg0) {
			log.info("[managedAccounts] " + arg0);
		}
		
		/**
		 * IB software calls this immediately after connecting to set reference orderID value.<br/>
		 * Sets the orderId(int) of the enclosing IBTwsConnection with the given id(int) and logs the following info:<br/>
		 * "[nextValidId] next valid id :${id}"
		 */
		public void nextValidId(int id) {
			//log.debug("[nextValidId] next valid id :" + id);
			orderId = id;
		}

		/**
		 * This method is the callback hook for all open orders, (open orders are orders that have been accepted at IB and are marked as open.)<br/>
		 * This is the state before execution. <br/>
		 * Iterates the ibEventListeners(Queue&lt;IBEventListener&gt;) of the enclosing IBTwsConnection and invokes
		 * <code>listener.openOrder(orderId, contract, order)</code> on each IBEventListener using the given orderId(int), contract(Contract) and order(Order)
		 */
		public void openOrder(int orderId, Contract contract, Order order, OrderState orderState) {
			// dispatch it. 
			for(IBEventListener listener : ibEventListeners){
				// fixme: 'whyHeld' information better be dispatched 				
				listener.openOrder(orderId, contract, order, orderState);
			}
		}
		/**
		 * <strong>1.</strong> Iterates the ibEventListeners(Queue&lt;IBEventListener&gt;) of the enclosing IBTwsConnection and invokes
		 * <code>listener.orderStatus(orderId, status, filled, remaining, avgFillPrice, permId, parentId, lastFillPrice, clientId)</code> on each IBEventListener using the given orderId(int), contract(Contract) and order(Order)<br/>
		 * <strong>2.</strong> Logs the following information:<br/>
		 * "***[orderStatus] ${orderId} / ${status} / ${filled} / ${remaining} / ${avgFillPrice} / ${permId} / ${parentId} / ${lastFillPrice} / ${clientId}/${whyHeld}"<br/>
		 * @param whyHeld can be "locate", "trigger" ..
		 * @see http://chuckcaplan.com/twsapi/index.php/void%20orderStatus%28%29
		 */
		public void orderStatus(int orderId, String status, int filled, int remaining, double avgFillPrice, int permId, int parentId, double lastFillPrice, int clientId, String whyHeld) {
			//log.info("[orderStatus] " + orderId + " / " + status + " / " + filled + " / " + remaining + " / " + avgFillPrice + " / " + permId + " / " + parentId + " / " + lastFillPrice + " / " + clientId + "/" + whyHeld);
		
			// dispatch it. 
			for(IBEventListener listener : ibEventListeners){
				listener.orderStatus(orderId, status, filled, remaining, avgFillPrice, permId, parentId, lastFillPrice, clientId, whyHeld);
			}
		}
		/**
		 * <strong>1.</strong> Gets (if there is one) the w(EWrapper) mapped under the given reqId(int) in the interceptors(Map&lt;Integer,EWrapper&gt;) and invokes
		 * <code>w.realtimeBar(reqId, time, open, high, low, close, volume, wap, count)</code> using the given reqId(int), time(long), open(double), high(double)
		 * , low(double), close(double), volume(long), wap(double) and count(int)<br/>
		 * <strong>2.</strong> Otherwise if there is no such mapping, it logs the following information:<br/>
		 * "[realtimeBar] orphan event: ${reqId}/${time}/${open}/${high}/${low}/${close}/${volume}/${wap}/${count}"
		 */
		public void realtimeBar(int reqId, long time, double open, double high, double low, double close, long volume, double wap, int count) {
			EWrapper w = interceptors.get(reqId);
			if(w != null) {
				w.realtimeBar(reqId, time, open, high, low, close, volume, wap, count);
				return;
			}
			log.warn("[realtimeBar] orphan event: " + reqId + "/" + time + "/" + open + "/" + high + "/" + low + "/" + close + "/" + volume + "/" + wap + "/" + count);
		}
		/**
		 * logs the following information:<br/>
		 * "[receiveFA] ${arg0}/ ${arg1}"
		 */
		public void receiveFA(int arg0, String arg1) {
			log.info("[receiveFA] " + arg0 + "/ " + arg1);
		}
		/**
		 * <strong>1.</strong> Gets (if there is one) the w(EWrapper) mapped under the given reqId(int) in the interceptors(Map&lt;Integer,EWrapper&gt;) and invokes
		 * <code>w.scannerData(reqId, rank, contractDetails, distance, benchmark, projection, legsStr)</code> using the given reqId(int), rank(int), contractDetails(ContractDetails),
		 * distance(String), benchmark(String), projection(String) and legsStr(String)<br/>
		 * <strong>2.</strong> Otherwise if there is no such mapping, it logs the following information:<br/>
		 * "[scannerData] orphan event: ${reqId}/${rank}/${contractDetails}/${distance}/${benchmark}/${projection}/${legsStr}"
		 */
		public void scannerData(int reqId, int rank, ContractDetails contractDetails, String distance, String benchmark, String projection, String legsStr) {
			log.debug("[scannerData]: " + reqId + "/" + rank + "/" + contractDetails + "/" + distance + "/" + benchmark + "/" + projection + "/" + legsStr);
			EWrapper w = interceptors.get(reqId);
			if(w != null) {
				w.scannerData(reqId, rank, contractDetails, distance, benchmark, projection, legsStr);
				return;
			}
			log.warn("[scannerData] orphan event: " + reqId + "/" + rank + "/" + contractDetails.m_summary.m_symbol + "/" + distance + "/" + benchmark + "/" + projection + "/" + legsStr);
		}
		/**
		 * logs the following information:<br/>
		 * "[scannerParameters] ${arg0}"
		 */
		public void scannerParameters(String arg0) {
			log.info("[scannerParameters] " + arg0);
		}
		/**
		 * <strong>1.</strong> Gets (if there is one) the w(EWrapper) mapped under the given reqId(int) in the interceptors(Map&lt;Integer,EWrapper&gt;) and invokes
		 * <code>w.tickEFP(reqId, tickType, basisPoints, formattedBasisPoints, impliedFuture, holdDays, futureExpiry, dividendImpact, dividendsToExpiry)</code> using the given reqId(int), 
		 * tickType(int), basisPoints(double), formattedBasisPoints(String), impliedFuture(double), holdDays(int), futureExpiry(String), dividendImpact(double) and dividendsToExpiry(double)<br/>
		 * <strong>2.</strong> Otherwise if there is no such mapping, it logs the following information:<br/>
		 * "[tickEFP] orphan event: ${reqId}/${tickType}/${basisPoints}/${formattedBasisPoints}/${impliedFuture}/${holdDays}/${futureExpiry}/${dividendImpact}/${dividendsToExpiry}"
		 */
		public void tickEFP(int reqId, int tickType, double basisPoints, String formattedBasisPoints, double impliedFuture, int holdDays, String futureExpiry, double dividendImpact, double dividendsToExpiry) {
			EWrapper w = interceptors.get(reqId);
			if(w != null) {
				w.tickEFP(reqId, tickType, basisPoints, formattedBasisPoints, impliedFuture, holdDays, futureExpiry, dividendImpact, dividendsToExpiry);
				return;
			}
			log.warn("[tickEFP] orphan event: " + reqId + "/" + tickType + "/" + basisPoints + "/" + formattedBasisPoints + "/" + impliedFuture + "/" + holdDays + "/" + futureExpiry + "/" + dividendImpact + "/" + dividendsToExpiry);
		}
		/**
		 * <strong>1.</strong> Gets (if there is one) the w(EWrapper) mapped under the given reqId(int) in the interceptors(Map&lt;Integer,EWrapper&gt;) and invokes
		 * <code>w.tickGeneric(reqId, tickType, value)</code> using the given reqId(int), tickType(int) and value(double)<br/>
		 * <strong>2.</strong> Otherwise if there is no such mapping, it logs the following information:<br/>
		 * "[tickGeneric] orphan event: ${reqId}/${tickType}/${value}"
		 */
		public void tickGeneric(int reqId, int tickType, double value) {
			EWrapper w = interceptors.get(reqId);
			if(w != null) {
				w.tickGeneric(reqId, tickType, value);
				return;
			}
			log.warn("[tickGeneric] orphan event: " + reqId + "/" + tickType + "/" + value);
		}
		/**
		 * <strong>1.</strong> Gets (if there is one) the w(EWrapper) mapped under the given reqId(int) in the interceptors(Map&lt;Integer,EWrapper&gt;) and invokes
		 * <code>w.tickOptionComputation(reqId, field, impliedVol, delta, optPrice, pvDividend, gamma, vega, theta, undPrice)</code> using the given reqId(int), field(int), impliedVol(double), delta(double) 
		 * optPrice(double) and pvDividend(double)<br/>
		 * <strong>2.</strong> Otherwise if there is no such mapping, it logs the following information:<br/>
		 * "[tickOptionComputation] orphan event: id=${reqId}"
		 */
		public void tickOptionComputation(int reqId, int field, double impliedVol, double delta, double optPrice, double pvDividend, double gamma, double vega, double theta, double undPrice) {
			log.debug("[tickOptionComputation] " + reqId + " / " + field + " / " + impliedVol + " / " + delta + "/" + optPrice + "/" + pvDividend + "/" + gamma + "/" + vega + "/" + theta + "/" + undPrice);
			EWrapper w = interceptors.get(reqId);
			if(w != null) {
				w.tickOptionComputation(reqId, field, impliedVol, delta, optPrice, pvDividend, gamma, vega, theta, undPrice);
				return;
			}
			log.warn("[tickOptionComputation] orphan event: id=" + reqId);
		}

		// this method updates server clock value
		// in my test it was off by 1 hour (probably because server does not
		// do DST while my computer does). It can be used to put better
		// timestamp on the received events.
		/**
		 * private final AtomicLong lastTimestamp = new AtomicLong();<br/>
		 * Used to store the last timeStamp received from the IB Server
		 */
		private final AtomicLong lastTimestamp = new AtomicLong();
		/**
		 * This method updates server clock value<br/>
		 * If the given tickType(int) is LAST_TIMESTAMP(45), it sets the associated lastTimestamp(AtomicLong) with the given value(String) (as long) and returns.<br/>
		 * Otherwise, it logs the following information:<br/>
		 * "[tickString] ${reqId}/${tickType}/${value}"<br/><br/>
		 * Note: in my test it was off by 1 hour (probably because server does not do DST while my computer does). It can be used to put better timestamp on the received events.
		 */
		public void tickString(int reqId, int tickType, String value) {
			if(tickType == TickType.LAST_TIMESTAMP) {
				try {
					long ts = Long.parseLong(value);
					lastTimestamp.set(ts);
//					log.info(new Date(ts * 1000));
				} catch(Exception ex) {
					log.warn(ex);
				}
				return;
			}
			
			log.info("[tickString] " + reqId + "/" + tickType + "/" + value);
		}
		/**
		 * <strong>1.</strong> Gets (if there is one) the w(EWrapper) mapped under the given tickId(int) in the interceptors(Map&lt;Integer,EWrapper&gt;) and invokes
		 * <code>w.tickPrice(tickId, tickType, price, canAutoExecute)</code> using the given tickId(int), tickType(int), price(double) and canAutoExecute(int)<br/>
		 * <strong>2.</strong> Otherwise if there is no such mapping, it logs the following information:<br/>
		 * "[tickPrice] orphan event: ${tickId}/${decodeTickType(tickType)}/${price}/${canAutoExecute}"
		 */
		public void tickPrice(int tickId, int tickType, double price, int canAutoExecute) {
			//log.info("[tickPrice] " + tickId + "/" + tickType + "/" + price+"/"+canAutoExecute);
			EWrapper w = interceptors.get(tickId);
			if(w != null) {
				w.tickPrice(tickId, tickType, price, canAutoExecute);
				return;
			}
			log.warn("[tickPrice] orphan event: " + tickId + "/" + decodeTickType(tickType) + "/" + price + "/" + canAutoExecute);
		}
		/**
		 * <strong>1.</strong> Gets (if there is one) the w(EWrapper) mapped under the given tickId(int) in the interceptors(Map&lt;Integer,EWrapper&gt;) and invokes
		 * <code>w.tickSize(tickId, tickType, size)</code> using the given tickId(int), tickType(int) and size(int)<br/>
		 * <strong>2.</strong> Otherwise if there is no such mapping, it logs the following information:<br/>
		 * "[tickSize] orphan event: ${tickId}/${decodeTickType(tickType)}/${size}"
		 */
		public void tickSize(int tickId, int tickType, int size) {
			EWrapper w = interceptors.get(tickId);
			if(w != null) {
				w.tickSize(tickId, tickType, size);
				return;
			}

			log.warn("[tickSize] orphan event: " + tickId + "/" + decodeTickType(tickType) + "/" + size);
		}
		/**
		 * logs the following information:<br/>
		 * "[updateAccountTime] ${arg0}"
		 */
		public void updateAccountTime(String arg0) {
			//log.info("[updateAccountTime] " + arg0);
		}
		/**
		 * <strong>1.</strong> Logs the following information:<br/>
		 * "[updateAccountValue] ${key}/ ${value} / ${currency} / ${accountName}"<br/>
		 * <strong>2.</strong> If the given key(String) and currency(String) are "CashBalance" and "USD", it iterates the ibEventListeners(Queue&lt;IBEventListener&gt;) of the enclosing IBTwsConnection and invokes
		 * <code>listener.updateAccountValue(key, value, currency, accountName)</code> on each IBEventListener using the given key(String), value(String), currency(String) and accountName(String)<br/>
		 * Other possible key values include:<br/>
		 * DayTradesRemaining - number of day trades left<br/>
		 * EquityWithLoanValue - equity with Loan Value<br/>
		 * InitMarginReq - current initial margin requirement<br/>
		 * MaintMarginReq - current maintenance margin<br/>
		 * NetLiquidation - net liquidation value<br/>
		 */
		public void updateAccountValue(String key, String value, String currency, String accountName) {
			//log.info("[updateAccountValue] " + key + "/ " + value + " / " + currency + " / " + accountName);
				if ("CashBalance".equals(key) && "USD".equals(currency)) {
					// dispatch it. 
					for(IBEventListener listener : ibEventListeners){
						listener.updateAccountValue(key, value, currency, accountName);
					}
				}				
		}
		/**
		 * <strong>1.</strong> Gets (if there is one) the w(EWrapper) mapped under the given reqId(int) in the interceptors(Map&lt;Integer,EWrapper&gt;) and invokes
		 * <code>w.updateMktDepth(reqId, position, operation, side, price, size)</code> using the given reqId(int), position(int), operation(int), side(int), price(double) and size(int)<br/>
		 * <strong>2.</strong> Otherwise if there is no such mapping, it logs the following information:<br/>
		 * "[updateMktDepth] orphan event: ${reqId}/${position}/${operation}/${side}/${price}/${size}"
		 */
		public void updateMktDepth(int reqId, int position, int operation, int side, double price, int size) {
			EWrapper w = interceptors.get(reqId);
			if(w != null) {
				w.updateMktDepth(reqId, position, operation, side, price, size);
				return;
			}

			log.warn("[updateMktDepth] orphan event: " + reqId + "/" + position + "/" + operation + "/" + side + "/" + price + "/" + size);
		}
		/**
		 * <strong>1.</strong> Gets (if there is one) the w(EWrapper) mapped under the given reqId(int) in the interceptors(Map&lt;Integer,EWrapper&gt;) and invokes
		 * <code>w.updateMktDepthL2(reqId, position, marketMaker, operation, side, price, size)</code> using the given reqId(int), position(int), marketMaker(String), operation(int), side(int), price(double) and size(int)<br/>
		 * <strong>2.</strong> Otherwise if there is no such mapping, it logs the following information:<br/>
		 * "[updateMktDepth2] orphan event: ${reqId}/${position}/${marketMaker}/${operation}/${side}/${price}/${size}"
		 */
		public void updateMktDepthL2(int reqId, int position, String marketMaker, int operation, int side, double price, int size) {
			EWrapper w = interceptors.get(reqId);
			if(w != null) {
				w.updateMktDepthL2(reqId, position, marketMaker, operation, side, price, size);
				return;
			}

			log.warn("[updateMktDepthL2] orphan event: " + reqId + "/" + position + "/" + marketMaker + "/" + operation + "/" + side + "/" + price + "/" + size);
		}
		/**
		 * this is a callback from reqNewsBulletin
		 */
		public void updateNewsBulletin(int msgId, int msgType, String message, String origExchange) {
			TimeStamp stamp=new TimeStamp();
			 IBNewsBulletin IBNewsBulletin=new IBNewsBulletin(msgId, msgType, stamp, message, origExchange);
			 log.info("[updateNewsBulletin]: "+IBNewsBulletin.toString());
//			EWrapper w = interceptors.get(reqId);
//			if(w != null) {
//				w.updateNewsBulletin(reqId, msgType, message, origExchange);
//				return;
//			}
//			log.warn("[updateNewsBulletin] orphan event: " + msgId + "/" + msgType + "/" + message + "/" + origExchange);
		}
		/**
		 * <strong>1.</strong> Logs the following information:<br/>
		 * "[updatePortfolio] ${contract}/ pos sum : ${positionCount} / ${marketPrice} / ${marketValue} / ${avgCost} / ${unrealizedPNL} / ${realizedPNL} / ${accountName}"<br/>
		 * <strong>2.</strong> Iterates the ibEventListeners(Queue&lt;IBEventListener&gt;) of the enclosing IBTwsConnection and invokes
		 * <code>listener.updatePortfolio(contract, positionCount, marketPrice, marketValue, avgCost, unrealizedPNL, realizedPNL, accountName)</code> on each IBEventListener using the 
		 * given contract(Contract), positionCount(int), marketPrice(double), marketValue(double), avgCost(double), unrealizedPNL(double), realizedPNL(double) and accountName(String)
		 */
		public void updatePortfolio(Contract contract, int positionCount, double marketPrice, double marketValue, double avgCost, double unrealizedPNL, double realizedPNL, String accountName) {
			//log.info("[updatePortfolio] " + contract.m_symbol + "/ pos sum : " + positionCount + " / mkt price:" + marketPrice + " / mkt value:" + marketValue + " / avg cost:" + avgCost + " / Unrlzd P&L:" + unrealizedPNL + " / P&L:" + realizedPNL + " / " + accountName);
			// dispatch it. 
			for(IBEventListener listener : ibEventListeners){
				listener.updatePortfolio(contract, positionCount, marketPrice, marketValue, avgCost, unrealizedPNL, realizedPNL, accountName);
			}
		}
		/**
		 * an empty method
		 */
		public void accountDownloadEnd(String accountName) {
			// TODO Auto-generated method stub
			
		}
		/**
		 * an empty method
		 */
		public void contractDetailsEnd(int reqId) {
			//log.info("[contractDetailsEnd] : " + reqId);
			EWrapper w = interceptors.get(reqId);
			if(w != null) {
				w.contractDetailsEnd(reqId);
				return;
			}
			log.warn("[contractDetailsEnd] orphan event: " + reqId);
			
		}
		/**
		 * an empty method
		 */
		public void deltaNeutralValidation(int reqId, UnderComp underComp) {
			// TODO Auto-generated method stub
			
		}
		/**
		 * an empty method
		 */
		public void fundamentalData(int reqId, String data) {
			log.debug("[fundamentalData] reqId="+reqId+" data="+data);
			
		}

		/**
		 * an empty method
		 */
		public void openOrderEnd() {
			log.debug("[openOrderEnd]");
			
		}
		/**
		 * 
		 */
		public void scannerDataEnd(int reqId) {
			//log.debug("[scannerDataEnd] event: " + reqId);
			EWrapper w = interceptors.get(reqId);
			if(w != null) {
				w.scannerDataEnd(reqId);
				return;
			}
			log.warn("[scannerDataEnd] orphan event: " + reqId);
			
		}

		public void tickSnapshotEnd(int reqId) {
			// TODO Auto-generated method stub
			
		}
	};
	/**
	 * Constructs an IBTwsConnection(extends Thread) using the given ip(String), port(int) and clientId(int) to set its associated ip(String), port(int) and clientId(int).<br/>
	 * It also sets its associated ecs(EClientSocket) to a new EClientSocket using the associated myWrapper(EWrapper) to set its m_anyWrapper(AnyWrapper implementation).
	 * Its EReader(extends Thread) is also a daemon Thread.
	 * @param ip
	 * @param port
	 * @param clientId
	 */
	public IBTwsConnection(String ip, int port, int clientId) {
		this.ecs = new EClientSocket(myWrapper) {
			
		    @Override
		    public EReader createReader(EClientSocket socket, DataInputStream dis) {
		    	EReader reader = super.createReader(socket, dis);
		    	reader.setDaemon(true);
		    	return reader;
		    }
		};
		
		this.ip = ip;
		this.port = port;
		this.clientId = clientId;
		log.info("[TwsApiMonitor] constructed.");
	}

	/**
	 * internal method to connect to TWS.<br/>
	 * <strong>1.</strong> If the associated connected(boolean) is false, it tries every 2 seconds to connect the associated ecs(EClientSocket) to TWS using the associated
	 * ip(String), port(int) and clientId(int). Doesn't do anything if the associated connected(boolean) is already true.<br/> 
	 * <strong>2.</strong> Once it connects, sets the associated connected(boolean) to true, and invokes 
	 * <code>ecs.reqExecutions(0,new ExecutionFilter())</code> and <code>ecs.reqAccountUpdates(true, "")</code> on the associated ecs(EClientSocket)
	 * to start receiving execution reports and account/portfolio updates
	 */
	public synchronized void connect() {
		if (!connected) {

			// doing a safety sleep for 2 seconds. 
			while(!connected){
				
				try{
					Thread.sleep(2000);
				} catch(Exception x) { }
				
				ecs.eConnect(ip, port, clientId);
				log.info("[connect] connected: " + ecs.isConnected());
				connected = ecs.isConnected();
				
				if(connected){
					int requestId = generateRequestId();
					ecs.reqExecutions(requestId,new ExecutionFilter());
					ecs.reqAccountUpdates(true, "");
					ecs.reqAllOpenOrders();
				}
			}
		}
	}

	/**
	 * internal method to disconnect from IB.<br/>
	 * <strong>1.</strong> Invokes <code>ecs.eDisconnect()</code> on the associated ecs(EClientSocket) which disconnects the connection to the TWS.<br/>
	 * <strong>2.</strong> Sets the associated connected(boolean) with the m_connected(boolean) of the associated ecs(EClientSocket)
	 *
	 */
	public void disconnect() {
		log.info("[disconnect]");
		ecs.eDisconnect();
		connected = ecs.isConnected();
	}

	/**
	 * internal method to trigger a reconnect. <br/>
	 * If there is no other reconnect in progress, it disconnects and connects to TWS, and then renews all existing data subscriptions
	 * from the associated tickWrappers(Map&lt;InstrumentSpecification,TickWrapper&gt;)
	 * 
	 */
	public void reconnect() {
		
			if(!reconnectInProgress){
				reconnectInProgress = true;
				disconnect();
				connect();
				// 
				resubscribe();
				reconnectInProgress = false;
			}
		
	}
	
	/**
	 * resubscribes / renews all existing data subscriptions after a reconnect.<br/>
	 * Required as the TWS does not continue streaming data after connection has been lost.<br/>
	 * Goes through each mapping in the associated tickWrappers(Map&lt;InstrumentSpecification,TickWrapper&gt;) and re-requests market data from the TWS for each InstrumentSpecification-TickWrapper pair.
	 * by invoking <code>ecs.reqMktData(w.id, contract, "", false)</code> on the associated ecs(EClientSocket) on each of them. <br/>
	 * It also renews the interceptro for each one of them in the associated interceptors(Map&lt;Integer,EWrapper&gt;).
	 */
	private void resubscribe(){
		//
		log.info("Resubscribing after reconnect ... ");
		Set<InstrumentSpecification> keys = tickWrappers.keySet();
		Iterator<InstrumentSpecification> iterator = keys.iterator();
		synchronized(tickWrappers){
		while(iterator.hasNext()){
			
			// specification 
			InstrumentSpecification spec = iterator.next();
			TickWrapper w = tickWrappers.get(spec);
			
			// remove the old interceptor. 
			interceptors.remove(w.id);
			w.id = generateRequestId();
			
			// add the interceptor again, this time with the new request id. 
			interceptors.put(w.id, w);
			
			// 
			Contract contract = convertToContract(spec);
			ecs.reqMktData(w.id, contract, "", false);
			log.info("requested market data for request id: " + w.id);
			
		}
		}
		
	}
	
	/**
	 * for tracing purposes<br/>
	 * returns the correct String for the given tickType(int):<br/>
	 * i.e.:<br/>
	 * BID_SIZE(0) -> "bidSize"<br/>
	 * BID(1) -> "bidPrice"<br/>
	 * ASK(2) -> "askPrice"<br/>
	 * ...
	 * SHORTABLE(46) -> "shortable"<br/>
	 * FUNDAMENTAL_RATIOS(47) -> "fundamentals"
	 */
	private static String decodeTickType(int type) {
		return TickType.getField(type);
	}
	/**
	 * private final AtomicInteger requestId = new AtomicInteger(0);
	 */
	private final AtomicInteger requestId = new AtomicInteger(0);
	/**
	 * increments the associated requestId(AtomicInteger) and returns its new value.
	 * @return
	 */
	private int generateRequestId() {
		return requestId.incrementAndGet();
	}
	
	/**
	 * Sets initial value for the request id. Useful when running two or more
	 * connections against single TWS: make sure that initial request ids are
	 * very different, so that there is no chance that they will interfere with
	 * each other.<br/>
	 * sets the value of the associated requestId(AtomicInteger) with the given val(int)
	 * @param val initial request id value.
	 */
	public void setRequestId(int val) {
		requestId.set(val);
	}
	
	/**
	 * class that can act as a listener for tick events (IB ticks are activequant's Quote or TradeIndication objects).<br/>
	 * A private static class. TickWrapper extends IBWrapper.<br/>
	 * Holds the following associated variables:
	 * <ul>
	 * <li>id(int)</li>
	 * <li>spec(InstrumentSpecification)</li>
	 * <li>quoteGenerator(UniqueDateGenerator)</li>
	 * <li>tradeGenerator(UniqueDateGenerator)</li>
	 * <li>quote(Quote)</li>
	 * <li>trade(TradeIndication)</li>
	 * <li>quoteEvent(Event&lt;Quote&gt;)</li>
	 * <li>tradeEvent(Event&lt;TradeIndication&gt;)</li>
	 * </ul>
	 */
	private static class TickWrapper extends IBWrapper {
		/**
		 * public int id;<br/>
		 * request id, can change after a reconnect. 
		 */
		public int id;
		/**
		 * private final InstrumentSpecification spec;<br/>
		 * request specification
		 */
		private final InstrumentSpecification spec;
		/**
		 * private final UniqueDateGenerator quoteGenerator = new UniqueDateGenerator();
		 */
		private final UniqueDateGenerator quoteGenerator = new UniqueDateGenerator(); 
		/**
		 * private final UniqueDateGenerator tradeGenerator = new UniqueDateGenerator();
		 */
		private final UniqueDateGenerator tradeGenerator = new UniqueDateGenerator(); 
		/**
		 * constructs a TickWrapper(extends IBWrapper) using the given id(int) and spec(InstrumentSpecification) to set its associated id(int) and spec(InstrumentSpecification)
		 * @param id
		 * @param spec
		 */
		TickWrapper(int id, InstrumentSpecification spec) {
			this.id = id;
			this.spec = spec;
		}
		/**
		 * private final Quote  quote = new Quote();<br/>
		 * accumulate quote parameters here
		 */
		private final Quote  quote = new Quote();
		/**
		 * private final TradeIndication trade = new TradeIndication();<br/>
		 * accumulate trade parameters here
		 */
		private final TradeIndication trade = new TradeIndication();
		/**
		 * public final Event&lt;Quote&gt; quoteEvent = new Event&lt;Quote&gt;();<br/>
		 * quote event dispatcher
		 */
		public final Event<Quote> quoteEvent = new Event<Quote>();
		/**
		 * public final Event&lt;TradeIndication&gt; tradeEvent = new Event&lt;TradeIndication&gt;();<br/>
		 * trade event dispatcher
		 */
		public final Event<TradeIndication> tradeEvent = new Event<TradeIndication>();
		/**
		 * <strong>1.</strong> Creates a clone(Quote) of the given quote(Quote) whose instrumentSpecification(InstrumentSpecification) and timeStamp(TimeStamp)
		 * are set with the associated spec(InstrumentSpecification) and current time.<br/>
		 * <strong>2.</strong> Fires that clone(Quote) on the associated quoteEvent(Event&lt;Quote&gt;) so any interested listers will get notified
		 * @param quote
		 */
		private void distributeQuote(Quote quote) {
			try {
				Quote q = quote.clone();
				
				q.setInstrumentSpecification(spec);
				q.setTimeStamp(quoteGenerator.generate(new Date()));
			
				quoteEvent.fire(q);
			} catch (Exception e) {
				e.printStackTrace();
				log.error(e);
			}
		}
		/**
		 * <strong>1.</strong> Creates a clone(TradeIndication) of the given trade(TradeIndication) whose instrumentSpecification(InstrumentSpecification) and timeStamp(TimeStamp)
		 * are set with the associated spec(InstrumentSpecification) and current time.<br/>
		 * <strong>2.</strong> Fires that clone(TradeIndication) on the associated tradeEvent(Event&lt;TradeIndication&gt;) so any interested listers will get notified
		 * @param trade
		 */
		private void distributeTrade(TradeIndication trade) {
			try {
				TradeIndication t = trade.clone();
				
				t.setInstrumentSpecification(spec);
				t.setTimeStamp(tradeGenerator.generate(new Date()));
			
				tradeEvent.fire(t);
			} catch (Exception e) {
				e.printStackTrace();
				log.error(e);
			}
		}

		/*
		 * the following logic related to "trade.setQuantity(0L);"
		 * tries to filter out duplicate tickSize()
		 * messages. This is because IB often generates duplicate tickSize()
		 * events, so that a trade of price 100. and size 3 may be reported as:
		 * 
		 * tickPrice(100.)
		 * tickSize(3)
		 * tickSize(3)
		 * 
		 * This happens not always, but approx in 50% events.
		 * 
		 * The logic below will distribute trade at first tickSize() event, but
		 * remember the size, and if next message is tickSize() one and
		 * tickSize() has the same exact size, it will be ignored.
		 */
		/**
		 * <strong>1.</strong> If the given tickType(int) is BID(1) or ASK(2), it sets the bidPrice(double) or askPrice(double) of the associated quote(Quote) with the given price(double)
		 * (and its bidQuantity or askQuantity to -1)<br/>
		 * <strong>2.</strong> If the given tickType(int) is LAST(4), it sets the price(double) of the associated trade(TradeIndication) with the given price(double).<br/>
		 * <strong>3.</strong> Otherwise, if the given tickType(int) is OPEN(14), LOW(7), HIGH(6), CLOSE(9) (or anything else), it just ignores it.
		 */
		@Override
		public void tickPrice(int tickId, int tickType, double price, int canAutoExecute) {
			//log.debug("[tick id=" + tickId + ": " + decodeTickType(tickType) + "/" + price + "/" + canAutoExecute);
			
			trade.setQuantity(0L);
			
			if (price <= 0.0) {
				// ignore obviously wrong thing
				// forex stream tend to start with ASK(0) BID(0) LAST(0) 
				return;
			}
				
			/**
			 * NOTE: IB software always calls tickSize() immediately after 
			 * tickPrice() for ticks of type BID/ASK/LAST (see source of their client). 
			 * Therefore I delay emitting the quote till we receive tickSize() call.
			 * Testing also shows that this pair is often accompanied by
			 * independent tickSize() event with the same parameters.
			 * This creates a possibility of generating duplicate events.
			 * To fight this duplication, BID/ASK events are generated only if
			 * event actually did change something (either price or size).
			 * This explains the logic below that says:
			 * if(price has changed) {
			 * 		force event distribution from tickSize() by setting size to -1
			 * }
			 * 
			 * We can not remove duplicates using this technique for LAST events,
			 * because its quite possible that two trades with same price and size 
			 * will occur in a row. This is still an unresolved problem.
			 * Presently, duplicate trade events are allowed to escape this code.
			 * 
			 * Also, I am researching the volume update events as a candidate for
			 * event generation point. Still work in progress.
			 * 
			 * -mpk
			 */
			switch (tickType) {
			case TickType.BID:
				if(price != quote.getBidPrice()) {
					quote.setBidPrice(price);
					quote.setBidQuantity(Quote.NOT_SET); // force tickSize() to emit this quote
				}
				break;
			case TickType.ASK:
				if(price != quote.getAskPrice()) {
					quote.setAskPrice(price);
					quote.setAskQuantity(Quote.NOT_SET); // force tickSize() to emit this quote
				}
				break;
			case TickType.LAST:
				trade.setPrice(price);
				break;
			case TickType.OPEN:
			case TickType.LOW:
			case TickType.HIGH:
			case TickType.CLOSE:
				log.debug("ignoring LOW/HIGH/CLOSE price event");
				break;
			default:
				log.warn("unhandled tick type: " + tickType);
//					throw new AssertionError("unexpected tick type: " + tickType);
			}				
		}
		/**
		 * <strong>1.</strong> If the given tickType(int) is BID_SIZE(0) or ASK_SIZE(3), it sets the bidQuantity(double) or askQuantity(double) of the associated quote(Quote) with the given size(int).
		 * If the Qute is complete(both bid/ask price/quantity are set), it distributes(fires) it.<br/>
		 * <strong>2.</strong> If the given tickType(int) is LAST_SIZE(5), it sets the quantity(double) of the associated trade(TradeIndication) with the given size(int).<br/>
		 * If its price(double) is already set, it distributes(fires) the trade(TradeIndication).<br/>
		 * <strong>3.</strong> Otherise, if the given tickType(int) is VOLUME(8) (or anything else), it ignores it
		 */
		@Override
		public void tickSize(int tickId, int tickType, int size) {
			//log.debug("[tick id=" + tickId + ": " + decodeTickType(tickType) + "/" + size);
			if(size == 0) {
				return; // obviously wrong thing: ignore
			}
			double lastSize = trade.getQuantity();
			trade.setQuantity(0L);			
			switch (tickType) {
			case TickType.ASK_SIZE:
				// do not flood with the same quotes!
				if(size != quote.getAskQuantity()) {
					quote.setAskQuantity(size);
					if(quote.getAskPrice() != Quote.NOT_SET) {
						// do not distribute incomplete quote
						if(quote.getBidQuantity() != Quote.NOT_SET ) {
							distributeQuote(quote);
						}
					} else {
						log.warn("orphan size message");
					}
				}
				break;

			case TickType.BID_SIZE:
				if(size != quote.getBidQuantity()) {
					quote.setBidQuantity(size);
					if(quote.getBidPrice() != Quote.NOT_SET) {
						// do not distribute incomplete quote
						if(quote.getAskQuantity() != Quote.NOT_SET ) {
							distributeQuote(quote);
						}
					} else {
						log.warn("orphan size message");
					}
				}
				break;
					
			case TickType.LAST_SIZE:
				if(trade.getPrice() != TradeIndication.NOT_SET) {
					if(lastSize != size) {
						trade.setQuantity(size);
						distributeTrade(trade);
					} else {
						log.debug("skipping duplicate message");
					}
				} else {
					log.warn("orphan size message");
				}
				break;
					
			case TickType.VOLUME:
				log.debug("ignoring volume update: " + size);
				break;
					
			default:
				log.warn("unhandled tick type: " + tickType);
//					throw new AssertionError("unexpected tick type");
			}
		}
		
		@Override
		public void error(int reqId, int errorCode, String message) {
			log.info("[error]: " + reqId + " / " + errorCode + " / " + message);
		}
	}
	
	/**
	 * class that can act as a listener for tick events (IB ticks are activequant's Quote or TradeIndication objects).<br/>
	 * A private static class. CandleWrapper extends IBWrapper. Holds the following associated variables:
	 * <ul>
	 * <li>id(int)</li>
	 * <li>spec(SeriesSpecification)</li>
	 * <li>quoteGenerator(UniqueDateGenerator)</li>
	 * <li>tradeGenerator(UniqueDateGenerator)</li>
	 * <li>candle(Candle)</li>
	 * <li>candleEvent(Event&lt;Candle&gt;)</li>
	 * </ul>
	 */
	private static class CandleWrapper extends IBWrapper {
		/**
		 * public int id;<br/>
		 * request id, can change after a reconnect. 
		 */
		public int id;
		/**
		 * private final SeriesSpecification spec;<br/>
		 * request specification
		 */
		private final SeriesSpecification spec;
		/**
		 * private final UniqueDateGenerator quoteGenerator = new UniqueDateGenerator(); 
		 */
		private final UniqueDateGenerator quoteGenerator = new UniqueDateGenerator(); 
		/**
		 * private final UniqueDateGenerator tradeGenerator = new UniqueDateGenerator(); 
		 */
		private final UniqueDateGenerator tradeGenerator = new UniqueDateGenerator(); 
		/**
		 * constructs a CandleWrapper(extends IBWrapper) using the given id(int) and spec(SeriesSpecification) to set its associated id(int) and spec(SeriesSpecification)
		 * @param id
		 * @param spec
		 */
		CandleWrapper(int id, SeriesSpecification spec) {
			this.id = id;
			this.spec = spec;
		}
		/**
		 * private final Candle candle = new Candle();
		 */
		private final Candle candle = new Candle();
		/**
		 * public final Event&lt;Candle&gt; candleEvent = new Event&lt;Candle&gt;();<br/>
		 * candle event dispatcher
		 */
		public final Event<Candle> candleEvent = new Event<Candle>();
		/**
		 * creates a clone Candle of the given candle(Candle) except its instrumentSpecification(InstrumentSpecification) and timeFrame(TimeFrame)
		 * are set with the ones of the associated spec(SeriesSpecification).<br/>
		 * Fires the clone Candle onto the associated candleEvent(Event&lt;Candle&gt;) so interested listeners will be notified.
		 * @param candle
		 */
		private void distributeCandle(Candle candle) {
			try {
				Candle c = candle.clone();
				
				c.setInstrumentSpecification(spec.getInstrumentSpecification());
				c.setTimeFrame(spec.getTimeFrame());
			
				candleEvent.fire(c);
			} catch (Exception e) {
				e.printStackTrace();
				log.error(e);
			}
		}
		/**
		 * WAP is not supported at the moment. <br/>
		 * <strong>1.</strong> Sets the openPrice(double), highPrice(double), lowPrice(double), closePrice(double), volume(double) and timeStamp(TimeStamp) of the associated 
		 * candle(Candle) with the given open(double), high(double), low(double), close(double), volume(double) and time(long). The other given variables aren't used.<br/>
		 * <strong>2.</strong> Creates a clone Candle of the associated candle(Candle) except its instrumentSpecification(InstrumentSpecification) and timeFrame(TimeFrame)
		 * are set with the ones of the associated spec(SeriesSpecification).<br/>
		 * Fires the clone Candle onto the associated candleEvent(Event&lt;Candle&gt;) so interested listeners will be notified.
		 */
		public void realtimeBar(int reqId, long time, double open, double high, double low, double close, long volume, double wap, int count) {
			candle.setOpenPrice(open);
			candle.setHighPrice(high);
			candle.setLowPrice(low);
			candle.setClosePrice(close);
			candle.setVolume(volume);
			candle.setWAP(wap);
			candle.setCount(count);
			candle.setTimeStamp(new TimeStamp(new Date(time*1000)));//seems like IB returns seconds, not milliseconds
			distributeCandle(candle);
		}

	}
	
	
	/**
	 * private final Map&lt;InstrumentSpecification,TickWrapper&gt; tickWrappers = new HashMap&lt;InstrumentSpecification,TickWrapper&gt;();
	 */
	private final Map<InstrumentSpecification,TickWrapper> tickWrappers = new HashMap<InstrumentSpecification,TickWrapper>();
	/**
	 * <strong>1.</strong> If the associated tickWrappers(Map&lt;InstrumentSpecification,TickWrapper&gt;) doesn't have a TickWrapper mapped under the given spec(InstrumentSpecification),
	 * it gets a unique requestId(int) and creates a TickWrapper using that requestId(int) and the given spec(InstrumentSpecification) to set its associated id(int) and spec(InstrumentSpecification).
	 * Then it maps it into the tickWrappers(Map&lt;InstrumentSpecification,TickWrapper&gt;). It also maps the requestId-TickWrapper pair into the associated interceptors(Map&lt;Integer,EWrapper&gt;)<br/>
	 * Then it connects to the TWS and invokes <code>ecs.reqMktData(requestId, contract, "", false)</code> to start getting market data for that contract.<br/>
	 * <strong>2.</strong> Adds the given listener(IEventListener&lt;TradeIndication&gt;) to the listeners(Queue&lt;IEventListener&lt;T&gt;&gt;) of the tradeEvent(Event&lt;TradeIndication&gt;)
	 * of the mapped (new or existing) TickWrapper
	 * @param listener
	 * @param spec
	 */
	public void subscribeToTradeIndication(final IEventListener<TradeIndication> listener, final InstrumentSpecification spec){
//		log.info("subscribe to TradeIndication stream: " + spec.getSymbol().toString()+" "+spec.getSecurityType().toString()+"@"+spec.getExchange());
	
		synchronized(tickWrappers) {
			TickWrapper w = tickWrappers.get(spec);
			if(w == null) {
				int requestId = generateRequestId();
				
				w = new TickWrapper(requestId, spec);
				tickWrappers.put(spec, w);
				interceptors.put(requestId, w);
				
				Contract contract = convertToContract(spec);

				connect();

				ecs.reqMktData(requestId, contract, "", false);
				//log.debug("reqested market data, request id=" + w.id);
			}

			w.tradeEvent.addEventListener(listener);
		}
	}
	/**
	 * <strong>1.</strong> If the associated tickWrappers(Map&lt;InstrumentSpecification,TickWrapper&gt;) doesn't have a TickWrapper mapped under the given spec(InstrumentSpecification),
	 * it returns without doing anything else.<br/>
	 * <strong>2.</strong> Removes the given listener(IEventListener&lt;TradeIndication&gt;) from the listeners(Queue&lt;IEventListener&lt;T&gt;&gt;) of the tradeEvent(Event&lt;TradeIndication&gt;)
	 * of the mapped w(TickWrapper). If both the tradeEvent(Event&lt;TradeIndication&gt;) and quoteEvent(Event&lt;Quote&gt;) have no more listeners, it notifies TWS to stop sending data for that 
	 * requestId by invoking <code>ecs.cancelMktData(w.id)</code> on the associated ecs(EClientSocket). It also removes the same TickWrapper from the associated interceptors(Map&lt;Integer,EWrapper&gt;)
	 * by invoking <code>interceptors.remove(w.id)</code>
	 * @param listener
	 * @param spec
	 */
	public void unsubscribeFromTradeIndication(IEventListener<TradeIndication> listener, InstrumentSpecification spec) {
//		log.info("unsubscribe from TradeIndication stream: " + spec.getSymbol().toString()+" "+spec.getSecurityType().toString()+"@"+spec.getExchange());

		synchronized(tickWrappers) {
			TickWrapper w = tickWrappers.get(spec);
			if(w == null) {
				log.warn("not subscribed");
				return;
			}

			w.tradeEvent.removeEventListener(listener);
			if(w.quoteEvent.isEmpty() && w.tradeEvent.isEmpty()) {
				ecs.cancelMktData(w.id);
				interceptors.remove(w.id);
				//log.debug("canceled market data, request id=" + w.id);
			}
		}
	}
	/**
	 * private final Map&lt;SeriesSpecification,CandleWrapper&gt; candleWrappers = new HashMap&lt;SeriesSpecification,CandleWrapper&gt;();
	 */
	private final Map<SeriesSpecification,CandleWrapper> candleWrappers = new HashMap<SeriesSpecification,CandleWrapper>();
	
	/**
	 * subscribes to a candle stream, this is a live feed of candles. <br/>
	 * <strong>1.</strong> If the associated candleWrappers(Map&lt;SeriesSpecification,CandleWrapper&gt;) doesn't have a CandleWrapper mapped under the given spec(SeriesSpecification),
	 * it gets a unique requestId(int) and creates a CandleWrapper using that requestId(int) and the given spec(SeriesSpecification) to set its associated id(int) and spec(SeriesSpecification).
	 * Then it maps it into the candleWrappers(Map&lt;SeriesSpecification,CandleWrapper&gt;). It also maps the requestId-CandleWrapper pair into the associated interceptors(Map&lt;Integer,EWrapper&gt;)<br/>
	 * Then it connects to the TWS and invokes <code>ecs.reqRealTimeBars(requestId, contract, barSize, type, false)</code> to start getting market data for that contract.<br/>
	 * <strong>2.</strong> Adds the given listener(IEventListener&lt;Candle&gt;) to the listeners(Queue&lt;IEventListener&lt;T&gt;&gt;) of the candleEvent(Event&lt;Candle&gt;)
	 * of the mapped (new or existing) CandleWrapper
	 * @param listener
	 * @param spec
	 * @param timeFrame
	 */
	public void subscribeToCandleStream(final IEventListener<Candle> listener, final SeriesSpecification spec){
//		log.info("subscribe to Candle stream for: " + spec );
	
		synchronized(candleWrappers) {
			CandleWrapper w = candleWrappers.get(spec);
			if(w == null) {
				int requestId = generateRequestId();
				
				w = new CandleWrapper(requestId, spec);
				candleWrappers.put(spec, w);
				interceptors.put(requestId, w);
				
				Contract contract = convertToContract(spec.getInstrumentSpecification());

				connect();
				int barSize = IBTwsConnection.timeFrameToBarSizeCode(spec.getTimeFrame());

				String type = TRADES;
				if(spec.getInstrumentSpecification().containsDecoration(DATA_TYPE)){
					type = spec.getInstrumentSpecification().getDecoration(DATA_TYPE);
				}
				
				ecs.reqRealTimeBars(requestId, contract, barSize, type, false);
				//log.info("reqested market data, request id=" + w.id);
			}

			w.candleEvent.addEventListener(listener);
		}
	}
	
	/**
	 * <strong>1.</strong> If the associated candleWrappers(Map&lt;SeriesSpecification,CandleWrapper&gt;) doesn't have a CandleWrapper mapped under the given spec(SeriesSpecification),
	 * it returns without doing anything else.<br/>
	 * <strong>2.</strong> Removes the given listener(IEventListener&lt;Candle&gt;) from the listeners(Queue&lt;IEventListener&lt;T&gt;&gt;) of the candleEvent(Event&lt;Candle&gt;)
	 * of the mapped w(CandleWrapper). If the candleEvent(Event&lt;Candle&gt;) has no more listeners, it notifies TWS to stop sending data for that 
	 * requestId by invoking <code>ecs.cancelRealTimeBars(w.id)</code> on the associated ecs(EClientSocket). It also removes the same CandleWrapper from the associated interceptors(Map&lt;Integer,EWrapper&gt;)
	 * by invoking <code>interceptors.remove(w.id)</code>
	 * @param listener
	 * @param spec
	 */
	public void unsubscribeFromCandleStream(IEventListener<Candle> listener, SeriesSpecification spec) {
//		log.info("unsubscribe from Candle stream: " + spec);

		synchronized(candleWrappers) {
			CandleWrapper w = candleWrappers.get(spec);
			if(w == null) {
				log.warn("not subscribed");
				return;
			}

			w.candleEvent.removeEventListener(listener);
			if(w.candleEvent.isEmpty()) {
				ecs.cancelRealTimeBars(w.id);
				interceptors.remove(w.id);
				//log.info("canceled candle market data for request id: " + w.id);
				// NOTE:added to make sure we get new request id's the next time we request market data!
				candleWrappers.remove(spec);
			}
		}
	}
	
	/**
	 * <strong>1.</strong> If the associated tickWrappers(Map&lt;InstrumentSpecification,TickWrapper&gt;) doesn't have a TickWrapper mapped under the given spec(InstrumentSpecification),
	 * it gets a unique requestId(int) and creates a TickWrapper using that requestId(int) and the given spec(InstrumentSpecification) to set its associated id(int) and spec(InstrumentSpecification).
	 * Then it maps it into the tickWrappers(Map&lt;InstrumentSpecification,TickWrapper&gt;). It also maps the requestId-TickWrapper pair into the associated interceptors(Map&lt;Integer,EWrapper&gt;)<br/>
	 * Then it connects to the TWS and invokes <code>ecs.reqMktData(requestId, contract, "", false)</code> to start getting market data for that contract.<br/>
	 * <strong>2.</strong> Adds the given listener(IEventListener&lt;Quote&gt;) to the listeners(Queue&lt;IEventListener&lt;T&gt;&gt;) of the quoteEvent(Event&lt;Quote&gt;)
	 * of the mapped (new or existing) TickWrapper
	 * @param listener
	 * @param spec
	 */
	public void subscribeToQuote(final IEventListener<Quote> listener, final InstrumentSpecification spec){
//		log.info("subscribe to Quote stream: " + spec.getSymbol().toString()+" "+spec.getSecurityType().toString()+"@"+spec.getExchange());
	
		synchronized(tickWrappers) {
			TickWrapper w = tickWrappers.get(spec);
			if(w == null) {
				int requestId = generateRequestId();
				
				w = new TickWrapper(requestId, spec);
				tickWrappers.put(spec, w);
				interceptors.put(requestId, w);
				
				Contract contract = convertToContract(spec);

				connect();
				
				IBHistoricalDataRequestEvent event=new IBHistoricalDataRequestEvent(requestId);
				try {
					historicRequestMemoryQueue.put(event);
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				event.setRequestSent(new TimeStamp());
				
				ecs.reqMktData(requestId, contract, "",false);
				//log.info("requested market data for request id: " + w.id);
			}

			w.quoteEvent.addEventListener(listener);
		}
	}
	/**
	 * <strong>1.</strong> If the associated tickWrappers(Map&lt;InstrumentSpecification,TickWrapper&gt;) doesn't have a TickWrapper mapped under the given spec(InstrumentSpecification),
	 * it returns without doing anything else.<br/>
	 * <strong>2.</strong> Removes the given listener(IEventListener&lt;Quote&gt;) from the listeners(Queue&lt;IEventListener&lt;T&gt;&gt;) of the quoteEvent(Event&lt;Quote&gt;)
	 * of the mapped w(TickWrapper). If both the tradeEvent(Event&lt;TradeIndication&gt;) and quoteEvent(Event&lt;Quote&gt;) have no more listeners, it notifies TWS to stop sending data for that 
	 * requestId by invoking <code>ecs.cancelMktData(w.id)</code> on the associated ecs(EClientSocket). It also removes the same TickWrapper from the associated interceptors(Map&lt;Integer,EWrapper&gt;)
	 * by invoking <code>interceptors.remove(w.id)</code>
	 * @param listener
	 * @param spec
	 */
	public void unsubscribeFromQuote(IEventListener<Quote> listener, InstrumentSpecification spec) {
//		log.info("unsubscribe from Quote stream: " + spec.getSymbol().toString()+" "+spec.getSecurityType().toString()+"@"+spec.getExchange());

		synchronized(tickWrappers) {
			TickWrapper w = tickWrappers.get(spec);
			if(w == null) {
				log.warn("not subscribed");
				return;
			}

			w.quoteEvent.removeEventListener(listener);
			if(w.quoteEvent.isEmpty() && w.tradeEvent.isEmpty()) {
				ecs.cancelMktData(w.id);
				interceptors.remove(w.id);
				//log.info("canceled market data for request id: " + w.id);
				// NOTE:added to make sure we get new request id's the next time we request market data!
				tickWrappers.remove(spec);
			}
		}
	}
	/**
	 * returns the associated connected(boolean)
	 * @return
	 */
	public boolean isConnected() {
		return connected;
	}
	/**
	 * returns the associated removeExpiredHistoricalRequests(boolean), set to true by default
	 * @return
	 */
	public boolean isRemoveExpiredHistoricalRequests() {
		return removeExpiredHistoricalRequests;
	}
	/**
	 * sets the associated removeExpiredHistoricalRequests(boolean) with the given removeExpiredHistoricalRequests(boolean)
	 * @param removeExpiredHistoricalRequests
	 */
	public void setRemoveExpiredHistoricalRequests(boolean removeExpiredHistoricalRequests) {
		this.removeExpiredHistoricalRequests = removeExpiredHistoricalRequests;
	}
	/**
	 * returns the associated requestAccountDataPeriodicaly(boolean), set to false by default
	 * @return
	 */
	public boolean isRequestAccountDataPeriodicaly() {
		return requestAccountDataPeriodicaly;
	}
	/**
	 * sets the associated requestAccountDataPeriodicaly(boolean) with the given requestAccountDataPeriodicaly(boolean)
	 * @param requestAccountDataPeriodicaly
	 */
	public void setRequestAccountDataPeriodicaly(boolean requestAccountDataPeriodicaly) {
		this.requestAccountDataPeriodicaly = requestAccountDataPeriodicaly;
	}
	/**
	 * If the associated removeExpiredHistoricalRequests(boolean) is set to true, it starts from the head of the associated
	 * historicRequestMemoryQueue(BlockingQueue&lt;IBHistoricalDataRequestEvent&gt;) and removes IBHistoricalDataRequestEvents until they are older
	 * than 12 Minutes
	 */
	private void removeExpiredHistoricalRequests() {
		if(!removeExpiredHistoricalRequests)return;
		TimeStamp now=new TimeStamp();
		TimeStamp TEN_MINS_AGO=new TimeStamp(now.getNanoseconds()-DURATION_TEN_MINUTES);
		while(!historicRequestMemoryQueue.isEmpty()&&historicRequestMemoryQueue.peek().getRequestSent().isBefore(TEN_MINS_AGO)){
			historicRequestMemoryQueue.poll();
		}
	}
	/**
	 * If the associated requestAccountDataPeriodicaly(boolean) is set to true, it sends a request to get the executions and account updates
	 */
	private void requestAccountDataPeriodicaly() {
		if(!requestAccountDataPeriodicaly)return;
		int requestId = generateRequestId();
		ecs.reqExecutions(requestId,new ExecutionFilter());
		ecs.reqAccountUpdates(true, "");
		//ecs.reqAllOpenOrders(); I only want to call this once, when I connect
	}
	/**
	 * required thread to keep this connection open and to fetch the list of executions once every so often ... <br/>
	 * This keeps connecting to TWS (in case we lost connection) every 5+25 secs and keeps requesting execution and account updates by invoking
	 * <code>ecs.reqExecutions(0,new ExecutionFilter())</code> and <code>ecs.reqAccountUpdates(true, "")</code> on the associated ecs(EClientSocket)
	 */
	public void run() {
		while (true) {
			connect();
			try {
				Thread.sleep(5000);
				// check if some orders are open ....
				if(this.connected) {
					requestAccountDataPeriodicaly();
					removeExpiredHistoricalRequests();
				}
				Thread.sleep(25000);
			} catch (Exception x) {
				log.warn("Exception while requesting executions", x);
			}
		}
	}
	
	/**
	 * helper method. <br/>
	 * returns an IB Contract by parsing the given symbol(String) which must be in the form:<br/>
	 * "${m_currency},${m_secType},${m_exchange},${m_primaryExch},${m_symbol},${m_expiry}/"
	 */
	public Contract getContractFromStringDefinition(String symbol) {
		StringTokenizer str = new StringTokenizer(symbol, ",");
		Contract contract = new Contract();
		contract.m_currency = str.nextToken().trim();
		contract.m_secType = str.nextToken().trim();
		contract.m_exchange = str.nextToken().trim();
		contract.m_primaryExch = contract.m_exchange;
		contract.m_symbol = str.nextToken().trim();
		contract.m_expiry = str.nextToken().trim();
		return contract;
	}
	
	/**
	 * helper method. <br/>
	 * returns a String representation of the given IB c(Contract) in the form:<br/>
	 * "${c.m_currency}/${c.m_secType}/${c.m_exchange}/${c.m_primaryExch}/${c.m_symbol}/${c.m_expiry}/"
	 */
	public String getContractString(Contract c){
		String ret = ""; 
		ret += c.m_currency +"/";
		ret += c.m_secType +"/";
		ret += c.m_exchange +"/";
		ret += c.m_primaryExch +"/";
		ret += c.m_symbol  +"/";
		ret += c.m_expiry  +"/";
		return ret; 
	}
	/**
	 * returns "IB"
	 * @return
	 */
	public String getVendorName() {
		return "IB";
	}
	/**
	 * returns the associated clientId(int)
	 * @return
	 */
	public int getClientId() {
		return clientId;
	}
	/**
	 * sets the associated clientId(int) with the given clientId(int)
	 * @param clientId
	 */
	public void setClientId(int clientId) {
		this.clientId = clientId;
	}
	/**
	 * returns the associated orderId(int)
	 * @return
	 */
	public int getOrderId() {
		return orderId;
	}
	/**
	 * increases the associated orderId(int) by 1 and then returns its value.
	 * @return
	 */
	public synchronized int getNextOrderId() {
		orderId++; 
		return orderId; 
	}

	/**
	 * adds an event listener, this is to be used only for IB specific events!!!<br/>
	 * Adds the given listener(IBEventListener) to the associated ibEventListeners(Queue&lt;IBEventListener&gt;) if its not already there.
	 * @param listener
	 */
	public void addIBEventListener(IBEventListener listener){
		if(!ibEventListeners.contains(listener))
			this.ibEventListeners.add(listener);
	}
	/**
	 * private final AtomicBoolean inFetch = new AtomicBoolean(false);<br/>
	 * prevents from two fetch requests running simultaneously<br/>
	 * @TODO: allow request queueing instead of throwing an exception
	 */
	private final AtomicBoolean inFetch = new AtomicBoolean(false);
	/**
	 * returns a CandleSeries holding IB hitorical data between the given startDate-endDate(TimeStamp) range, for the given spec(SeriesSpecification),
	 * barSize(String) (e.g. "1 second", "1 minute" etc), method(String)<br/>
	 * and method(String) (valid values: TRADES, MIDPOINT, BID, ASK, BID_ASK, HISTORICAL_VOLATILITY, OPTION_IMPLIED_VOLATILITY, OPTION_VOLUME).<br/>
	 * It gets all data (not only one within Regular Trading Hours).<br/>
	 * Data is fetched by fetchGranularity(String) (5 DAY by default) intervals at a time, and there is 8 seconds wait between each call (IB rule).<br/>
	 * It also has logic that prevents it from doing more than one request at a time.
	 * @param startDate
	 * @param endDate
	 * @param barSize
	 * @param method
	 * @param spec
	 * @return
	 * @throws Exception
	 */
	public CandleSeries fetch(TimeStamp startDate, TimeStamp endDate, String barSize, String method, SeriesSpecification spec, boolean useRTH) throws Exception {
		
		if(startDate == null) {
			throw new NullPointerException("startDate");
		}
		
		if(endDate == null) {
			throw new NullPointerException("endDate");
		}
		
		if(method == null) {
			throw new NullPointerException("method");
		}

		if(spec == null) {
			throw new NullPointerException("spec");
		}
		
		if(inFetch.getAndSet(true)) {
			throw new IllegalStateException("another fetch request is still running. IB allows only one fetch at a time");
		}
		String prevFetchGranularity=fetchGranularity;
		optimizeFetchGranularityFetchGranularity(spec.getTimeFrame());
		try {
			return doFetch(startDate.getDate(), endDate.getDate(), barSize, method, spec, useRTH);
		} finally {
			if(optimizeFetchGranularity){
				setFetchGranularity(prevFetchGranularity);
			}
			inFetch.set(false);
		}
	}
	
	/**
	 * returns a QuoteSeries holding IB hitorical data between the given startDate-endDate(TimeStamp) range, for the given spec(SeriesSpecification),
	 * barSize(String) (e.g. "1 second", "1 minute" etc), method(String)<br/>
	 * and method(String) (valid values: TRADES, MIDPOINT, BID, ASK, BID_ASK, HISTORICAL_VOLATILITY, OPTION_IMPLIED_VOLATILITY, OPTION_VOLUME).<br/>
	 * It gets all data (not only one within Regular Trading Hours).<br/>
	 * Data is fetched by fetchGranularity(String) (5 DAY by default) intervals at a time, and there is 8 seconds wait between each call (IB rule).<br/>
	 * It also has logic that prevents it from doing more than one request at a time.
	 * @param startDate
	 * @param endDate
	 * @param barSize
	 * @param method
	 * @param spec
	 * @return
	 * @throws Exception
	 */
	public QuoteSeries fetchQuotes(TimeStamp startDate, TimeStamp endDate, String barSize, String method, SeriesSpecification spec) throws Exception {
		
		if(startDate == null) {
			throw new NullPointerException("startDate");
		}
		
		if(endDate == null) {
			throw new NullPointerException("endDate");
		}
		
		if(method == null) {
			throw new NullPointerException("method");
		}

		if(spec == null) {
			throw new NullPointerException("spec");
		}
		
		if(inFetch.getAndSet(true)) {
			throw new IllegalStateException("another fetch request is still running. IB allows only one fetch at a time");
		}
		String prevFetchGranularity=fetchGranularity;
		optimizeFetchGranularityFetchGranularity(spec.getTimeFrame());
		try {
			return doFetchQuotes(startDate.getDate(), endDate.getDate(), barSize, method, spec);
		} finally {
			if(optimizeFetchGranularity){
				setFetchGranularity(prevFetchGranularity);
			}
			inFetch.set(false);
		}
	}
	
	/**
	 * private final AtomicBoolean isCandleFetched = new AtomicBoolean(false);
	 */
	private final AtomicBoolean isCandleFetched = new AtomicBoolean(false);
	/**
	 * private final AtomicBoolean isQuoteFetched = new AtomicBoolean(false);
	 */
	private final AtomicBoolean isQuoteFetched = new AtomicBoolean(false);
	/**
	 * private final AtomicReference&lt;CandleSeries&gt; candleSeriesBag = new AtomicReference&lt;CandleSeries&gt;();
	 */
	private final AtomicReference<CandleSeries> candleSeriesBag = new AtomicReference<CandleSeries>();
	/**
	 * private final AtomicReference&lt;QuoteSeries&gt; quoteSeriesBag = new AtomicReference&lt;QuoteSeries&gt;();
	 */
	private final AtomicReference<QuoteSeries> quoteSeriesBag = new AtomicReference<QuoteSeries>();
	/**
	 * private static final int DURATION_SECOND = 1;
	 */
	private static final int DURATION_SECOND = 1;
	/**
	 * private static final long DURATION_TEN_MINUTES = 10*60*DURATION_SECOND*1000000000L;
	 */
	private static final long DURATION_TEN_MINUTES = 10*60*DURATION_SECOND*1000000000L;
	/**
	 * private static final int DURATION_DAY = 24*60*60*DURATION_SECOND; (24*60*60)
	 */
	public static final int DURATION_DAY    = 24 * 60 * 60 * DURATION_SECOND;
	/**
	 * private static final int DURATION_WEEK   = 7 * DURATION_DAY; (7*24*60*60)
	 */
	private static final int DURATION_WEEK   = 7 * DURATION_DAY;
	/**
	 * private static final int DURATION_YEAR   = 365 * DURATION_DAY; (365*24*60*60)
	 */
	private static final int DURATION_YEAR   = 365 * DURATION_DAY;
	/**
	 * private int duration = 5 * DURATION_DAY;
	 */
	private int duration = 5 * DURATION_DAY;	
	/**
	 * private String fetchGranularity = "5 D";
	 */
	private String fetchGranularity = "5 D";
	
	/**
	 * Fetch granularity. This is a parameter that is specific to how historic data
	 * is implemented in IB. Their interface allow to request historical data
	 * ending at a particular date and having a particular duration. The duration
	 * values are limited. Therefore, long time spans are split in multiple requests
	 * each one fetching only for the duration, until all time span is covered.
	 * This duration is called <code>fetchGranularity</code>.
	 * Known values that are accepted are:<br/>
	 * "1 D" (one day), "1 W" (one week), "3600 S" (one hour).<br/>
	 * returns the associated fetchGranularity(String)
	 * @return fetch granularity string.
	 */
	public String getFetchGranularity() {
		return fetchGranularity;
	}
	
	/**
	 * Sets fetch granularity string. May throw an exception if invalid value is passed.<br/>
	 * Sets the associated fetchGranularity(String) with the given val(String)<br/>
	 * It needs to be something like "${x} D", "${x} W" or "${x} S" etc.<br/>
	 * It also sets the associated duration(int) with the equivalent of the given period in seconds.
	 * @param val fetch granularity string.
	 */
	public void setFetchGranularity(String val) {
		String [] vv = val.split(" ");
		if(vv.length != 2) {
			throw new IllegalArgumentException("failed to parse granularity.");
		}
		int value = Integer.parseInt(vv[0]);
		if("Y".equals(vv[1])) {
			duration = value * DURATION_YEAR;//1 Y is max
		} else if("D".equals(vv[1])) {
			duration = value * DURATION_DAY;
		} else if("W".equals(vv[1])) {
			duration = value * DURATION_WEEK;
		} else if("S".equals(vv[1])) {
			duration = value * DURATION_SECOND;
		} else {
			throw new IllegalArgumentException("failed to parse units");
		}
		fetchGranularity = val;
	}

	/**
	 * private final EWrapper histCandleCallback = new IBWrapper() {...}<br/>
	 * callback to receive historic events.
	 */
	private final EWrapper histCandleCallback = new IBWrapper() {
		/**
		 * This is the callback which gets called after we request historical data from IB.<br/>
		 * Here it uses the received data to create a Candle and adds it to the beggining of the current CandleSeries in the associated candleSeriesBag(AtomicReference&lt;CandleSeries&gt;).
		 * It keeps doing the same until it sees a negative given open(double) which means there is no more data. Then it sets the isCandleFetched(AtomicBoolean) to true and notifies the waitiong threads
		 * it is done.
		 */
		@Override
		public void historicalData(int reqId, String date, double open, double high, double low, double close, int volume, int count, double WAP, boolean hasGaps) {
			log.debug("[historicalData] " + reqId + " / " + date + " / " + open + " / " + high + " / " + low + " / " + close + " / " + volume + " / " + count + " / " + WAP + "/" + hasGaps);

			if (open >= 0.0) {
				try {
					// checking the format found in the data set.
					TimeStampFormat tsf = (date.length()== 8) ? new TimeStampFormat("yyyyMMdd") : new TimeStampFormat("yyyyMMdd HH:mm:ss");
//					tsf.setTimeZone(TimeZone.getDefault()); // FIXME: which timezone IB uses?
					
					tsf.setTimeZone(NASDAQCalendarSettings.getInstance().getTimeZone());
					// constructing a new candle.
					Candle c = new Candle();
					c.setOpenPrice(open);
					c.setHighPrice(high);
					c.setLowPrice(low);
					c.setClosePrice(close);
					c.setVolume(volume);
					c.setTimeStamp(tsf.parse(date));
					c.setCount(count);
					c.setWAP(WAP);
					c.setHasGaps(hasGaps);
					// get the current timeseries.
					CandleSeries ts = candleSeriesBag.get();
					ts.add(0, c);
				} catch (Exception x) {
					log.warn(StackTraceParser.getStackTraceMessage(x.getStackTrace()));
				}
			} else {
				// finished
				isCandleFetched.set(true);
				synchronized(isCandleFetched) {
					isCandleFetched.notifyAll();
				}
			}
		}

		@Override
		public void error(int reqId, int errorCode, String message) {
			log.info("[error]: " + reqId + " / " + errorCode + " / " + message);
				
			switch(errorCode){
			case 162:
			case 200:
			case 203:
				isCandleFetched.set(true);
				synchronized(isCandleFetched) {
					isCandleFetched.notifyAll();
				}
				break;

			default:
				log.error("unhandled error: code=" + errorCode);
			}
		}
	};
	
	/**
	 * private final EWrapper histQuoteCallback = new IBWrapper() {...}<br/>
	 * callback to receive historic events.
	 */
	private final EWrapper histQuoteCallback = new IBWrapper() {
		/**
		 * This is the callback which gets called after we request historical data from IB.<br/>
		 * Here it uses the received data to create a Quote and adds it to the beggining of the current QuoteSeries in the associated quoteSeriesBag(AtomicReference&lt;QuoteSeries&gt;).
		 * It keeps doing the same until it sees a negative given open(double) which means there is no more data. Then it sets the isQuoteFetched(AtomicBoolean) to true and notifies the waitiong threads
		 * it is done.
		 */
		@Override
		public void historicalData(int reqId, String date, double open, double high, double low, double close, int volume, int count, double WAP, boolean hasGaps) {
			log.debug("[historiclaData] " + reqId + " / " + date + " / " + open + " / " + high + " / " + low + " / " + close + " / " + volume + " / " + count + " / " + WAP + "/" + hasGaps);
			//Random randomSizeGenerator = new Random();//temporary
			if (open >= 0.0) {
				try {
					// checking the format found in the data set.
					TimeStampFormat tsf = (date.length()== 8) ? new TimeStampFormat("yyyyMMdd") : new TimeStampFormat("yyyyMMdd HH:mm:ss");
//					tsf.setTimeZone(TimeZone.getDefault()); // FIXME: which timezone IB uses?
//					tsf.setTimeZone(NASDAQCalendarSettings.getInstance().getTimeZone());
					tsf.setTimeZone(TimeZone.getTimeZone("GMT-5:00"));
					// constructing a new candle.
					Quote c = new Quote();
					c.setBidPrice(open);
					c.setAskPrice(close);
					c.setBidQuantity(-1);
					c.setAskQuantity(-1);

					c.setTimeStamp(tsf.parse(date));
					// get the current timeseries.
					QuoteSeries ts = quoteSeriesBag.get();
					ts.add(0, c);
				} catch (Exception x) {
					log.warn(StackTraceParser.getStackTraceMessage(x.getStackTrace()));
				}
			} else {
				// finished
				isQuoteFetched.set(true);
				synchronized(isQuoteFetched) {
					isQuoteFetched.notifyAll();
				}
			}
		}

		@Override
		public void error(int reqId, int errorCode, String message) {
			log.info("[error]: " + reqId + " / " + errorCode + " / " + message);
				
			switch(errorCode){
			case 162:
			case 200:
			case 203:
				isQuoteFetched.set(true);
				synchronized(isQuoteFetched) {
					isQuoteFetched.notifyAll();
				}
				break;

			default:
				log.error("unhandled error: code=" + errorCode);
			}
		}
	};
	
	/**
	 * private final static long PACING_DELAY = 8000;<br/>
	 * IB requires to sleep at least 8 seconds before backfill requests :(
	 */
	private final static long PACING_DELAY = 1000;//8000;
	/**
	 * private final AtomicLong lastFetch = new AtomicLong(0L);
	 */
	private final AtomicLong lastFetch = new AtomicLong(0L);
	/**
	 * if less than PACING_DELAY(8 secs) have passed since the time in milliseconds set in the associated lastFetch(AtomicLong),
	 * it causes the current thread to sleep for the rest of those 8 seconds.
	 * @throws InterruptedException
	 */
	private void doPacingSleep() throws InterruptedException {
		long old = lastFetch.get();
		if(old == 0L) return; // no need to sleep: first call
		
		long diff = PACING_DELAY - (System.currentTimeMillis() - old);
		if(diff > 0) Thread.sleep(diff);
	}
	/**
	 * private boolean stopFetchingOnEmpty = false;
	 */
	private boolean stopFetchingOnEmpty = false;
	
	/**
	 * returns the associated stopFetchingOnEmpty(boolean)<br/>
	 * This parameter controls internal logic of candle fetching.
	 * The default value if <code>false</code>. When fetching candles,
	 * requested time interval is automatically split into sub-intervals
	 * (see {@link #fetchGranularity} parameter). Most recent sub-interval is
	 * requested first, then goes next most recent, etc. So we move into past.
	 * It may happen that IB historic database does not go that far into the past.
	 * Yet, we would continue, by default, requesting more and more distant past intervals,
	 * until we exhaust the full range requested by user.
	 * <p>
	 * By setting this parameter to <code>true</code>, the iteration will cease
	 * as soon as first sub-interval returns zero candles.
	 * <p>
	 * Be careful with it, as incorrect setting of other parameters may
	 * confuse algorithm into terminating too early. For example, if {@link #fetchGranularity}
	 * is set to 3 days, and we hit a holiday, happened to be Friday, this three-day
	 * interval will not return any candles and will cause the algorithm 
	 * to stop fetching candles. This will result in less candles returned than IB has.
	 * 
	 * @return flag value.
	 */
	public boolean isStopFetchingOnEmpty() { 
		return stopFetchingOnEmpty;
	}
	
	/**
	 * sets the associated stopFetchingOnEmpty(boolean) with the given val(boolean)<br/>
	 * See {@link #isStopFetchingOnEmpty()}.
	 * @param val
	 */
	public void setStopFetchingOnEmpty(boolean val) {
		stopFetchingOnEmpty = val;
	}
	/**
	 * returns a CandleSeries holding IB hitorical data between the given startDate-endDate(Date) range, for the given query(SeriesSpecification),
	 * barSize(String) (e.g. "1 second", "1 minute" etc), and method(String) <br/>
	 * (valid values: TRADES, MIDPOINT, BID, ASK, BID_ASK, HISTORICAL_VOLATILITY, OPTION_IMPLIED_VOLATILITY, OPTION_VOLUME).<br/>
	 * It gets all data (not only one within Regular Trading Hours).<br/>
	 * Data is fetched by fetchGranularity(String) (5 DAY by default) intervals at a time, and there is 8 seconds wait between each call (IB rule)
	 * @param startDate
	 * @param endDate
	 * @param barSize
	 * @param method
	 * @param query
	 * @return
	 * @throws Exception
	 */
	private CandleSeries doFetch(Date startDate, Date endDate, String barSize, String method, SeriesSpecification query, boolean useRTH) throws Exception {
		connect();
		//log.info("[fetch] about to fetch : " + query);
		//log.info("[fetch] Data type: " + method);
		// build a new IB contract object;
		Contract contract = convertToContract(query.getInstrumentSpecification());
		SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd HH:mm:ss zzz");
		sdf.setTimeZone(TimeZone.getTimeZone("GMT"));
//		log.info("[fetch] Fetching  " + getContractString(contract) + " / " + barSize + " / " + method);
		// calculate the amount of days between start and end date.
		CandleSeries tempSeries = new CandleSeries(query);
		Calendar endCal = GregorianCalendar.getInstance(TimeZone.getTimeZone("GMT"));
		endCal.setTimeInMillis(endDate.getTime());
		while (endCal.getTime().compareTo(startDate) > 0) {
			//log.info("[fetch] iterating.");
			String startString = sdf.format(endCal.getTime());
			//log.info("[fetch] about to fetch a tranche " + startString);
			// fetch it.
			// building candle series with series specification
			CandleSeries cs = new CandleSeries();
			cs.setSeriesSpecification(query);
			candleSeriesBag.set(cs);
			int tickId = generateRequestId();
			interceptors.put(tickId, histCandleCallback);
			doPacingSleep();
			//log.info("put into map: " + tickId + ", " + histCandleCallback);			
			//log.info("[reqHistoricalData: " + tickId + "/" + contract + "/" + startString + "/" + fetchGranularity + "/" + barSize + "/" + method);
			
			//try {
			    IBHistoricalDataRequestEvent event=new IBHistoricalDataRequestEvent(tickId);
				historicRequestMemoryQueue.put(event);
				event.setRequestSent(new TimeStamp());
			//} catch (InterruptedException e) {
			//	throw new RuntimeException(e);
			//}
			ecs.reqHistoricalData(tickId, contract, startString, fetchGranularity, barSize, method, useRTH?1:0, 1);
			isCandleFetched.set(false);
			synchronized(isCandleFetched) {
				try { 
					isCandleFetched.wait(360 * 1000);
				} catch(InterruptedException ex) {
				}
			}
			if(!isCandleFetched.get()) {
				log.warn("timeout waiting for candles");
			}

			interceptors.remove(tickId);
			//log.info("removed from map: " + tickId);

			//log.debug("[fetch] tranche fetched. About to merge. This may take a while ... Please be patient. ");
			// merge it with the temporary time series object.
			CandleSeries tempSeries2 = candleSeriesBag.get();
			
			if(stopFetchingOnEmpty && tempSeries2.size() == 0) {
				log.info("fetch loop interrupted as interval returned no candles (assume there is no point in going further)");
				break;
			}
			
			//log.info("merging: base.size=" + tempSeries.size() + ", incoming.size=" + tempSeries2.size());
			tempSeries = TimeSeriesUtils.merge(tempSeries, tempSeries2);

//			log.info("[fetch] timeseries merged. result.size=" + tempSeries.size());

			lastFetch.set(System.currentTimeMillis());
			endCal.add(Calendar.SECOND, -duration);
		}
			
		//log.info("[fetch] CandleSeries fetched.");

		for (Candle c : tempSeries) {
			c.setHighTimeStamp(c.getTimeStamp());
			c.setLowTimeStamp(c.getTimeStamp());
			c.setTimeFrame(query.getTimeFrame());
			c.setInstrumentSpecification(query.getInstrumentSpecification());
		}

		// setting the series specification.  
		tempSeries.setSeriesSpecification(query);
		
		// disconnect
		return tempSeries;
	}
	
	/**
	 * returns a QuoteSeries holding IB hitorical data between the given startDate-endDate(Date) range, for the given query(SeriesSpecification),
	 * barSize(String) (e.g. "1 second", "1 minute" etc), , and method(String) <br/>
	 * (valid values: TRADES, MIDPOINT, BID, ASK, BID_ASK, HISTORICAL_VOLATILITY, OPTION_IMPLIED_VOLATILITY, OPTION_VOLUME).<br/>
	 * It gets all data (not only one within Regular Trading Hours).<br/>
	 * Data is fetched by fetchGranularity(String) (5 DAY by default) intervals at a time, and there is 8 seconds wait between each call (IB rule)
	 * @param startDate
	 * @param endDate
	 * @param barSize
	 * @param method
	 * @param query
	 * @return
	 * @throws Exception
	 */
	private QuoteSeries doFetchQuotes(Date startDate, Date endDate, String barSize, String method, SeriesSpecification query) throws Exception {
		connect();
		log.info("[fetch] about to fetch : " + query);
		//log.info("[fetch] Data type: " + method);
		// build a new IB contract object;
		Contract contract = convertToContract(query.getInstrumentSpecification());
		//System.out.println("BarSize " + barSize);		
		SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd HH:mm:ss zzz");
		sdf.setTimeZone(TimeZone.getTimeZone("GMT"));
		//log.info("[fetch] Fetching  " + getContractString(contract) + " / " + barSize + " / " + method);
		// calculate the amount of days between start and end date.
		QuoteSeries tempSeries = new QuoteSeries(query);
		Calendar endCal = GregorianCalendar.getInstance(TimeZone.getTimeZone("GMT"));
		endCal.setTimeInMillis(endDate.getTime());
		while (endCal.getTime().compareTo(startDate) > 0) {
			//log.info("[fetch] iterating.");
			String startString = sdf.format(endCal.getTime());
//			log.info("[fetch] about to fetch a tranche " + startString);
			// fetch it.
			// building candle series with series specification
			QuoteSeries cs = new QuoteSeries();
			cs.setSeriesSpecification(query);
			quoteSeriesBag.set(cs);
			int tickId = generateRequestId();
			interceptors.put(tickId, histQuoteCallback);
			doPacingSleep();
			//log.info("put into map: " + tickId + ", " + histQuoteCallback);			
			//log.info("[reqHistoricalData: " + tickId + "/" + contract + "/" + startString + "/" + fetchGranularity + "/" + barSize + "/" + method);
			//try {
			IBHistoricalDataRequestEvent event=new IBHistoricalDataRequestEvent(tickId);
			historicRequestMemoryQueue.put(event);
			event.setRequestSent(new TimeStamp());
			//} catch (InterruptedException e) {
			//	throw new RuntimeException(e);
			//}
			ecs.reqHistoricalData(tickId, contract, startString, fetchGranularity, barSize, method, 0, 1);
			isQuoteFetched.set(false);
			synchronized(isQuoteFetched) {
				try { 
					isQuoteFetched.wait(360 * 1000);
				} catch(InterruptedException ex) {
				}
			}
			if(!isQuoteFetched.get()) {
				log.error("timeout waiting for quotes");
			}

			interceptors.remove(tickId);
			//log.info("removed from map: " + tickId);

			//log.info("[fetch] tranche fetched. About to merge. This may take a while ... Please be patient. ");
			// merge it with the temporary time series object.
			QuoteSeries tempSeries2 = quoteSeriesBag.get();
			
			if(stopFetchingOnEmpty && tempSeries2.size() == 0) {
				log.info("fetch loop interrupted as interval returned no quotes (assume there is no point in going further)");
				break;
			}
			
			//log.info("merging: base.size=" + tempSeries.size() + ", incoming.size=" + tempSeries2.size());
			tempSeries = TimeSeriesUtils.merge(tempSeries, tempSeries2);

//			log.info("[fetch] timeseries merged. result.size=" + tempSeries.size());

			lastFetch.set(System.currentTimeMillis());
			endCal.add(Calendar.SECOND, -duration);
		}
			
		log.info("[fetch] QuoteSeries fetched.");

		for (Quote c : tempSeries) {
			c.setInstrumentSpecification(query.getInstrumentSpecification());
		}

		// setting the series specification.  
		tempSeries.setSeriesSpecification(query);
		
		// disconnect
		return tempSeries;
	}
	/**
	 * returns an IB Contract by converting the given spec(InstrumentSpecification).<br/>
	 * The Contract will have a currency, secType, exchange, primaryExchange, symbol and optionally:
	 * expiry, strike and right.
	 * @param spec
	 * @return
	 */
	private static Contract convertToContract(InstrumentSpecification spec){
		SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
		sdf.setTimeZone(TimeZone.getTimeZone("UTC"));
		com.ib.client.Contract newContract = new com.ib.client.Contract();
		
		if(spec.getCurrency()==null)throw new IllegalArgumentException("Missing currency in spec");
		
		newContract.m_currency = spec.getCurrency().name();
		if(spec.getSecurityType()==null)throw new IllegalArgumentException("Missing security type in spec");
		newContract.m_secType = fromSecurityTypeEnum(spec.getSecurityType());
		if(spec.getExchange()==null)throw new IllegalArgumentException("Missing exchange in spec");
		newContract.m_exchange = spec.getExchange();
		newContract.m_primaryExch = spec.getPrimaryExchange();// no aggregate exchange like SMART
		if(spec.getSymbol()==null)throw new IllegalArgumentException("Missing symbol in spec");
		newContract.m_symbol = spec.getSymbol().toString();
		if(spec.getExpiry() != null){
			newContract.m_expiry = sdf.format(spec.getExpiry().getTimeStamp().getDate());
		}
		if(spec.hasStrike()){
			newContract.m_strike = spec.getStrike();
		}
		if(spec.hasContractRight()){
			newContract.m_right = spec.getContractRight();
		}
		return newContract;
	}
	/**
	 * returns an InstrumentSpecification by converting the given IB contract(Contract) into an InstrumentSpecification.<br/>
	 * The InstrumentSpecification will have a currency, symbol, securityType, and optionally: strike, expiry, exchange and contractRight
	 * @param contract
	 * @return
	 * @throws Exception
	 */
	public static InstrumentSpecification convertToInstrument(Contract contract) throws Exception {
		SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
		sdf.setTimeZone(TimeZone.getTimeZone("UTC"));
		InstrumentSpecification spec = new InstrumentSpecification();
		
		String cur = contract.m_currency;
		spec.setCurrency(Currency.valueOf(cur));
		
		String symbol = contract.m_symbol;
		spec.setSymbol(new Symbol(symbol));
		
		if(contract.m_strike!=0.0){
			spec.setStrike(contract.m_strike);
		}
		
		spec.setSecurityType(toSecurityTypeEnum(contract.m_secType));
		
		// the expiry
		if(contract.m_expiry != null){
			spec.setExpiry(new Expiry(new TimeStamp(sdf.parse(contract.m_expiry))));
		}
		
		if(contract.m_primaryExch != null){
			spec.setPrimaryExchange(contract.m_primaryExch);
		}
		
		if(contract.m_exchange != null){
			spec.setExchange(contract.m_exchange);
		}else{
			spec.setExchange("SMART");
		}
		
		if(contract.m_right != null&&!"0".equals(contract.m_right)&&!"?".equals(contract.m_right)){
			spec.setContractRight(contract.m_right); //openOrder returns "0" and updatePortfolio returns "?"
		}
		spec.setVendor("IB");
		return spec;
	}
	/**
	 * returns the correct SecurityType enum for the given type(String):<br/>
	 * "FUT" -> SecurityType.FUTURE<br/>
	 * "FOP" -> SecurityType.FUTURE_OPTION<br/>
	 * "IND" -> SecurityType.INDEX<br/>
	 * "OPT" -> SecurityType.OPTION<br/>
	 * "STK" -> SecurityType.STOCK<br/>
	 * "CASH" -> SecurityType.CASH
	 * @param type
	 * @return
	 */
	public static SecurityType toSecurityTypeEnum(String type) {
		if(type.equals("FUT"))
			return SecurityType.FUTURE;
		else if(type.equals("FOP"))
			return SecurityType.FUTURE_OPTION;
		else if(type.equals("IND"))
			return SecurityType.INDEX;
		else if(type.equals("OPT"))
			return SecurityType.OPTION;
		else if(type.equals("STK"))
			return SecurityType.STOCK;
		else if(type.equals("CASH"))
			return SecurityType.CASH;
		else if(type.equals("BAG"))
			return SecurityType.BAG;
		throw new IllegalArgumentException("Unknown security type");
	}
	/**
	 * returns the correct IB String for the given type(SecurityType) enum:<br/>
	 * SecurityType.FUTURE -> "FUT"<br/>
	 * SecurityType.FUTURE_OPTION -> "FOP"<br/>
	 * SecurityType.INDEX -> "IND"<br/>
	 * SecurityType.OPTION -> "OPT"<br/>
	 * SecurityType.STOCK -> "STK"<br/>
	 * SecurityType.CASH -> "CASH"
	 * @param type
	 * @return
	 */
	public static String fromSecurityTypeEnum(SecurityType type) {
		if(type == SecurityType.FUTURE)
			return "FUT";
		else if(type == SecurityType.FUTURE_OPTION)
			return "FOP";
		else if(type == SecurityType.INDEX)
			return "IND";
		else if(type == SecurityType.OPTION)
			return "OPT";
		else if(type == SecurityType.STOCK)
			return "STK";
		else if(type == SecurityType.CASH)
			return "CASH";
		else if(type == SecurityType.BAG)
			return "BAG";
		throw new IllegalArgumentException("Unknown security type");
	}
	/**
	 * returns a String (human readable) representation of the given timeFrame(TimeFrame). It can be something like:<br/>
	 * "${x} second(seconds)"<br/>
	 * "${x} minunte(minutes)"<br/>
	 * "1 min"<br/>
	 * "${x} hour"<br/>
	 * "${x} day"<br/>
	 * "${x} week"<br/>
	 * "${x} months"<br/>
	 * "1 month"<br/>
	 * "${x} year"
	 * @param timeFrame
	 * @return
	 */
	public static String timeFrameToBarSize(TimeFrame timeFrame) {
		String out = Integer.toString(timeFrame.length);
		switch (timeFrame.unit) {
		case SECOND: 
			return out + " secs";
		case MINUTE:
			if(timeFrame.length>1)
				return out + " mins";
			else
				return out + " min";
		case HOUR:
			return out + " hour";
		case DAY:
			return out + " day";
		case WEEK:
			return out + " week";
		case MONTH:
			if(timeFrame.length>1)
				return out + " months";
			else 
				return out + " month";
		case YEAR:
			return out + " year";
		}
		
		throw new IllegalArgumentException("unsupported time frame: " + timeFrame);
	}
	/**
	 * returns an IB barSizeCode(int) equivalent for the given timeFrame(TimeFrame):<br/>
	 * <strong>1 SECOND -> 1</strong><br/>
	 * <strong>5 SECOND -> 2</strong><br/>
	 * <strong>15 SECOND -> 3</strong><br/>
	 * <strong>30 SECOND -> 4</strong><br/>
	 * <strong>1 MINUTE -> 5</strong><br/>
	 * <strong>2 MINUTE -> 6</strong><br/>
	 * <strong>5 MINUTE -> 7</strong><br/>
	 * <strong>15 MINUTE -> 8</strong><br/>
	 * <strong>30 MINUTE -> 9</strong><br/>
	 * <strong>1 HOUR -> 10</strong><br/>
	 * <strong>1 DAY -> 11</strong><br/>
	 * <strong>1 WEEK -> 12</strong><br/>
	 * <strong>1 MONTH -> 13</strong><br/>
	 * <strong>3 MONTH -> 14</strong><br/>
	 * <strong>1 YEAR -> 15</strong>
	 * @param timeFrame
	 * @return
	 */
	public static int timeFrameToBarSizeCode(TimeFrame timeFrame) {
		switch (timeFrame.unit) {
		case SECOND:
			if(timeFrame.length == 1) {
				return 1;
			} else if(timeFrame.length == 5) {
				return 2;
			} else if(timeFrame.length == 15) {
				return 3;
			} else if(timeFrame.length == 30) {
				return 4;
			} 
			break;
		case MINUTE:     
			if(timeFrame.length == 1) {
				return 5;
			} else if(timeFrame.length == 2) {
				return 6;
			} else if(timeFrame.length == 5) {
				return 7;
			} else if(timeFrame.length == 15) {
				return 8;
			} else if(timeFrame.length == 30) {
				return 9;
			}
			break;
		case HOUR:
			if(timeFrame.length == 1) {
				return 10;
			}
			break;
		case DAY:
			if(timeFrame.length == 1) {
				return 11;
			}
			break;
		case WEEK:
			if(timeFrame.length == 1) {
				return 12;
			}
			break;
		case MONTH:
			if(timeFrame.length == 1) {
				return 13;
			} else if(timeFrame.length == 3) {
				return 14;
			}
			break;
		case YEAR:
			if(timeFrame.length == 1) {
				return 15;
			}
			break;
		}
		
		throw new IllegalArgumentException("unsupported time frame: " + timeFrame);
	}
	
/* TODO
	// Broker methods
	private com.ib.client.Order convertToTwsOrder(org.activequant.tradesystem.domainmodel.Order o){
		// need to convert the order into an IB order object.

		com.ib.client.Order twsOrder = new com.ib.client.Order();
		// get the next implementation specific order id and set it in the order
		// object.
		int orderId = getNextOrderId(); // generate next valid order id
		o.setOriginalId((long) orderId);

		twsOrder.m_orderId = orderId;
		twsOrder.m_clientId = getClientId();
		twsOrder.m_transmit = true;
		
		switch(o.getSide()) {
		case BUY:
			twsOrder.m_action = "BUY";
			break;
		case SELL:
			twsOrder.m_action = "SELL";
			break;
		case SHORT_SELL: case SHORT_SELL_EXEMPT:
			twsOrder.m_action = "SSHORT";
			break;
		default:
			throw new IllegalArgumentException("unsupported order side: " + o);
		}

		// set the quantity. IB supports only integers.
		twsOrder.m_totalQuantity = (int) o.getQuantity();

		switch(o.getType()) {
		case MARKET:
			twsOrder.m_orderType = "MKT";
			break;
			
		case LIMIT:
			twsOrder.m_orderType = "LMT";
			twsOrder.m_lmtPrice = o.getLimit();
			break;
			
		case STOP:
			twsOrder.m_orderType = "STP";
			twsOrder.m_auxPrice = o.getStop();
			break;
			
		case STOP_LIMIT:
			twsOrder.m_orderType = "STPLMT";
			twsOrder.m_lmtPrice = o.getLimit();
			break;
			
		default:
			throw new IllegalArgumentException("unsupported order type: " + o);
		}
		
		return twsOrder;
	}
*/
	/**
	 * Instructs TWS to place the given IB twsOrder(Order) for the given spec(InstrumentSpecification) by invoking
	 * <code>ecs.placeOrder(twsOrder.m_orderId, convertToContract(spec), twsOrder)</code> on the associated ecs(EClientSocket)
	 */
	public void placeOrder(InstrumentSpecification spec, com.ib.client.Order twsOrder) throws Exception {
		log.info("placing tws order " + twsOrder.m_orderId);
		ecs.placeOrder(twsOrder.m_orderId, convertToContract(spec), twsOrder);
	}
	/**
	 * instructs TWS to cancel the Order with the given twsOrderId(int) by invoking
	 * <code>ecs.cancelOrder(twsOrderId)</code> on the associated ecs(EClientSocket)
	 * @param twsOrderId
	 * @throws Exception
	 */
	public void cancelOrder(int twsOrderId) throws Exception {
		log.info("canceling tws order " + twsOrderId);
		ecs.cancelOrder(twsOrderId);
	}
	/**
	 * If the given allMsgs(boolean) is set to TRUE, returns all the existing bulletins for the current day and any new ones. IF set to FALSE, will only return new bulletins.	
	 * @param allMsgs
	 * @throws Exception
	 */
	public void requestNewsBulletins(boolean allMsgs) throws Exception {
		log.info("requesting News Bulletins " + allMsgs);
		ecs.reqNewsBulletins(allMsgs);
	}
	
	public void cancelNewsBulletins() throws Exception {
		log.info("canceling News Bulletins");
		ecs.cancelNewsBulletins();
	}
	
	/**
	 * when set, fetch granularity for historuical data will be calculated on the fly, and optimized so it gets  near the max 2000 results per call.
	 */
	private boolean optimizeFetchGranularity=false;
	
	public boolean isOptimizeFetchGranularity() {
		return optimizeFetchGranularity;
	}

	public void setOptimizeFetchGranularity(boolean optimizeFetchGranularity) {
		this.optimizeFetchGranularity = optimizeFetchGranularity;
	}
	
	private void optimizeFetchGranularityFetchGranularity(TimeFrame timeFrame){
		if(!inFetch.get()) {
			throw new IllegalStateException("fetchGranularity optimization is only allowed within a historic fetch request.");
		}
		if(optimizeFetchGranularity){
			if(timeFrame==TimeFrame.TIMEFRAME_1_MINUTE){
				setFetchGranularity("6 D");
			}else if(timeFrame==TimeFrame.TIMEFRAME_60_MINUTES){
				setFetchGranularity("31 D"); //34 D system limit, lets use 31 for 1 month
			}else if(timeFrame==TimeFrame.TIMEFRAME_1_DAY){
				setFetchGranularity("1 Y");
			}			
		}
	}
	/**
	 * private final Map&lt;ScanCriteria,MarketScannerWrapper&gt; marketScanWrappers = new HashMap&lt;ScanCriteria,MarketScannerWrapper&gt;();
	 */
	private final Map<ScanCriteria,MarketScannerWrapper> marketScanWrappers = new HashMap<ScanCriteria,MarketScannerWrapper>();
	
	public void subscribeToMarketScanStream(final IEventListener<MarketScan> listener, final ScanCriteria spec){
		
		synchronized(marketScanWrappers) {
			MarketScannerWrapper w = marketScanWrappers.get(spec);
			if(w == null) {
				int requestId = generateRequestId();
				
				w = new MarketScannerWrapper(requestId, spec);
				marketScanWrappers.put(spec, w);
				interceptors.put(requestId, w);
				
				ScannerSubscription subscription= ScannerUtil.convertToScannerSubscription(spec);
				connect();
			
				ecs.reqScannerSubscription(requestId, subscription);
				log.info("subscribe to MarketScanner stream for: " + spec.getScanCode() +" request id=" + w.id);
				//log.info("reqested MarketScanner data, request id=" + w.id);
			}
			w.marketScanEvent.addEventListener(listener);
		}
	}
	
	public void unsubscribeFromMarketScanStream(IEventListener<MarketScan> listener, ScanCriteria spec) {
		log.info("unsubscribe from MarketScanner stream: " + spec.getScanCode());

		synchronized(marketScanWrappers) {
			MarketScannerWrapper w = marketScanWrappers.get(spec);
			if(w == null) {
				log.warn("not subscribed");
				return;
			}

			w.marketScanEvent.removeEventListener(listener);
			if(w.marketScanEvent.isEmpty()) {
				ecs.cancelScannerSubscription(w.id);
				interceptors.remove(w.id);
				//log.info("cancel MarketScanner data for request id: " + w.id);
				// NOTE:added to make sure we get new request id's the next time we request market data!
				marketScanWrappers.remove(spec);
			}
		}
	}
	
	private static class MarketScannerWrapper extends IBWrapper {
		/**
		 * public int id;<br/>
		 * request id, can change after a reconnect. 
		 */
		public int id;
		/**
		 * private final ScanCriteria spec;<br/>
		 * request specification
		 */
		private final ScanCriteria spec;
		/**
		 * private final UniqueDateGenerator scanGenerator = new UniqueDateGenerator(); 
		 */
		private final UniqueDateGenerator scanGenerator = new UniqueDateGenerator(); 
		/**
		 * constructs a CandleWrapper(extends IBWrapper) using the given id(int) and spec(SeriesSpecification) to set its associated id(int) and spec(SeriesSpecification)
		 * @param id
		 * @param spec
		 */
		MarketScannerWrapper(int id, ScanCriteria spec) {
			this.id = id;
			this.spec = spec;
			this.marketScan = new MarketScan(spec);
		}

		private final MarketScan marketScan;
		/**
		 * public final Event&lt;Candle&gt; candleEvent = new Event&lt;Candle&gt;();<br/>
		 * candle event dispatcher
		 */
		public final Event<MarketScan> marketScanEvent = new Event<MarketScan>();
		/**
		 * creates a clone Candle of the given candle(Candle) except its instrumentSpecification(InstrumentSpecification) and timeFrame(TimeFrame)
		 * are set with the ones of the associated spec(SeriesSpecification).<br/>
		 * Fires the clone Candle onto the associated candleEvent(Event&lt;Candle&gt;) so interested listeners will be notified.
		 * @param candle
		 */
		private void distributeMarketScan(MarketScan marketScan) {
			try {
				MarketScan c = marketScan.clone();
//				c.setScanCriteria(spec);
				//c.setInstrumentSpecification(spec.getInstrumentSpecification());
				//c.setTimeFrame(spec.getTimeFrame());
				c.setTimeStamp(scanGenerator.generate(new Date()));
				marketScanEvent.fire(c);
			} catch (Exception e) {
				e.printStackTrace();
				log.error(e);
			}
		}
		@Override
		public void scannerDataEnd(int reqId) {
			if(this.id!=reqId){
				throw new IllegalArgumentException(" wrong request id:"+reqId+". Was expecting reqId="+this.id);
			}
			if(!marketScan.getRows().isEmpty())
			{ 
			  distributeMarketScan(marketScan);
			  marketScan.reset();
			}
			
		}
		@Override
		public void scannerData(int reqId, int rank, ContractDetails contractDetails, String distance, String benchmark, String projection, String legsStr) {
			if(this.id!=reqId){
				throw new IllegalArgumentException(" wrong request id:"+reqId+". Was expecting reqId="+this.id);
			}

			InstrumentSpecificationDetails specDetails;
			try {
				specDetails = ScannerUtil.convertToInstrumentSpecificationDetails(contractDetails);
				specDetails.setTimeStamp(scanGenerator.generate(new Date()));
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
				throw new RuntimeException("We can't have this");
			}
			marketScan.insertMarketScanRow(rank, specDetails, distance, benchmark, projection, legsStr);
		}
		
		@Override
		public void error(int reqId, int errorCode, String message) {
			log.info("[error]: " + reqId + " / " + errorCode + " / " + message);
		}
	}
	
	private class InstrumentSpecificationWrapper extends IBWrapper {
		/**
		 * public int id;<br/>
		 * request id, can change after a reconnect. 
		 */
		public int id;
		/**
		 * private final InstrumentSpecification spec;<br/>
		 * request specification
		 */
		private final InstrumentSpecification spec;
		
		private ArrayList<InstrumentSpecificationDetails> instrumentSpecificationDetailsList;
		/**
		 * private final UniqueDateGenerator dateGenerator = new UniqueDateGenerator(); 
		 */
		private final UniqueDateGenerator dateGenerator = new UniqueDateGenerator(); 
		/**
		 * constructs a CandleWrapper(extends IBWrapper) using the given id(int) and spec(InstrumentSpecification) to set its associated id(int) and spec(InstrumentSpecification)
		 * @param id
		 * @param spec
		 */
		InstrumentSpecificationWrapper(int id, InstrumentSpecification spec) {
			this.id = id;
			this.spec = spec;
		}
		
		/**
		 * public final Event&lt;InstrumentSpecificationDetails&gt; specDetailsEvent = new Event&lt;InstrumentSpecificationDetails&gt;();<br/>
		 * candle event dispatcher
		 */
		public final Event<ArrayList<InstrumentSpecificationDetails>> specDetailsEvent = new Event<ArrayList<InstrumentSpecificationDetails>>();
		/**
		 * creates a clone InstrumentSpecificationDetails of the given detailsList(ArrayList&lt;InstrumentSpecificationDetails&gt;)<br/>
		 * Fires the clone InstrumentSpecificationDetails onto the associated specDetailsEvent(Event&lt;InstrumentSpecificationDetails&gt;) so interested listeners will be notified.
		 * @param candle
		 */
		private void distributeInstrumentSpecificationDetails(ArrayList<InstrumentSpecificationDetails> detailsList) {
			try {
				if (detailsList.size() == 1) {
//					if (spec.hasId()) {
//						detailsList.get(0).getInstrumentSpecification().setId(spec.getId());
//					}
					if (spec.getExchange() != null && detailsList.get(0).getInstrumentSpecification().getExchange() == null) {
						detailsList.get(0).getInstrumentSpecification().setExchange(spec.getExchange());
					}
					// details.setInstrumentSpecification(spec);
				}
				TimeStamp timestamp = dateGenerator.generate(new Date());
				for(InstrumentSpecificationDetails details:detailsList){
					details.setTimeStamp(timestamp);
				}
				specDetailsEvent.fire(detailsList);
			} catch (Exception e) {
				e.printStackTrace();
				log.error(e);
			}
		}
		
		@Override
		public void contractDetailsEnd(int reqId) {
			if(this.id!=reqId){
				throw new IllegalArgumentException(" wrong request id:"+reqId+". Was expecting reqId="+this.id);
			}
			distributeInstrumentSpecificationDetails(instrumentSpecificationDetailsList);
			EWrapper w=interceptors.get(reqId);
			if(w instanceof InstrumentSpecificationWrapper){
				InstrumentSpecificationWrapper isw=(InstrumentSpecificationWrapper)w;
				isw.specDetailsEvent.clear();
				interceptors.remove(reqId);
			} else{
				throw new RuntimeException("[contractDetailsEnd] We can't have this");
			}
			// finished
			isSpecDetailsFetched.set(true);
			synchronized(isSpecDetailsFetched) {
				isSpecDetailsFetched.notifyAll();
			}
			
		}
		@Override
		public void contractDetails(int reqId, ContractDetails contractDetails) {
			if(this.id!=reqId){
				throw new IllegalArgumentException(" wrong request id:"+reqId+". Was expecting reqId="+this.id);
			}
			//Contract contract = convertToContract(spec.getInstrumentSpecification());
			try {
				if(instrumentSpecificationDetailsList==null){
					instrumentSpecificationDetailsList=new ArrayList<InstrumentSpecificationDetails>();
				}
				InstrumentSpecificationDetails instrumentSpecificationDetails = ScannerUtil.convertToInstrumentSpecificationDetails(contractDetails);
				instrumentSpecificationDetailsList.add(instrumentSpecificationDetails);
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
				throw new RuntimeException("[contractDetails] We can't have this");
			}
		}
		
		@Override
		public void error(int reqId, int errorCode, String message) {
			log.info("[error]: " + reqId + " / " + errorCode + " / " + message);
		}
	}
	//private final AtomicBoolean inRequest = new AtomicBoolean(false);
	private final AtomicBoolean isSpecDetailsFetched = new AtomicBoolean(false);
	public void requestInstrumentSpecificationDetails(final IEventListener<ArrayList<InstrumentSpecificationDetails>> listener, final InstrumentSpecification spec) {
//		if (inRequest.getAndSet(true)) {
//			throw new IllegalStateException("another request for spec details is still running. IB allows only one request at a time");
//		}
		try {
			synchronized (interceptors) {
				int requestId = generateRequestId();
				InstrumentSpecificationWrapper w = new InstrumentSpecificationWrapper(requestId, spec);
				interceptors.put(requestId, w);
				Contract contract = convertToContract(spec);
				connect();
				w.specDetailsEvent.addEventListener(listener);
				isSpecDetailsFetched.set(false);
				ecs.reqContractDetails(requestId, contract);
				log.info("reqested Contract Details data, request id=" + w.id + " for symbol:" + contract.m_symbol);
				
				synchronized(isSpecDetailsFetched) {
					try { 
						isSpecDetailsFetched.wait(360 * 1000);
					} catch(InterruptedException ex) {
					}
				}
				if(!isSpecDetailsFetched.get()) {
					log.error("timeout waiting for quotes");
				}
			}
		} finally {
			//inRequest.set(false);
		}
	}
	
	private static class MarketDepthWrapper extends IBWrapper {
		/**
		 * public int id;<br/>
		 * request id, can change after a reconnect. 
		 */
		public int id;
		/**
		 * private final InstrumentSpecification spec;<br/>
		 * request specification
		 */
		private final InstrumentSpecification spec;
		/**
		 * private final UniqueDateGenerator dateGenerator = new UniqueDateGenerator(); 
		 */
		private final UniqueDateGenerator dateGenerator = new UniqueDateGenerator(); 
		/**
		 * constructs a MarketDepthWrapper(extends IBWrapper) using the given id(int) and spec(InstrumentSpecification) to set its associated id(int), spec(InstrumentSpecification) and marketDepth(MarketDepth)
		 * @param id
		 * @param spec
		 */
		MarketDepthWrapper(int id, InstrumentSpecification spec) {
			this.id = id;
			this.spec = spec;
//			this.marketDepth = new MarketDepth(spec);
			this.marketDepthEventHolder = new MarketDepthEvent(spec);
		}
		/**
		 * private final MarketDepth marketDepth;
		 */
//		private final MarketDepth marketDepth;
		
		private final MarketDepthEvent marketDepthEventHolder;
		
		public final Event<MarketDepthEvent> marketDepthEvent = new Event<MarketDepthEvent>();

		private void distributeMarketDepthEvent(MarketDepthEvent event) {
			try {
				MarketDepthEvent c = event.clone();
//				c.setInstrumentSpecification(spec);
				c.setTimeStamp(dateGenerator.generate(new Date()));
				marketDepthEvent.fire(c);
			} catch (Exception e) {
				e.printStackTrace();
				log.error(e);
			}
		}
		
		/**
		 * public final Event&lt;MarketDepth&gt; marketDepthEvent = new Event&lt;MarketDepth&gt;();<br/>
		 * marketDepth event dispatcher
		 */
//		public final Event<MarketDepth> marketDepthEvent = new Event<MarketDepth>();
//
//		private void distributeMarketDepth(MarketDepth marketDepth) {
//			try {
//				MarketDepth c = marketDepth.clone();
//				c.setInstrumentSpecification(spec);
//				c.setTimeStamp(dateGenerator.generate(new Date()));
//				marketDepthEvent.fire(c);
//			} catch (Exception e) {
//				e.printStackTrace();
//				log.error(e);
//			}
//		}
		
		@Override
		public void updateMktDepth(int reqId, int position, int operation, int side, double price, int size) {
			if(this.id!=reqId){
				throw new IllegalArgumentException("wrong request id:"+reqId+". Was expecting reqId="+this.id);
			}
			marketDepthEventHolder.setMarketMaker("");
			marketDepthEventHolder.setOperation(operation);
			marketDepthEventHolder.setPosition(position);
			marketDepthEventHolder.setPrice(price);
			marketDepthEventHolder.setSide(side);
			marketDepthEventHolder.setSize(size);
			distributeMarketDepthEvent(marketDepthEventHolder);
//			marketDepth.updateMktDepth(position, "", operation, side, price, size);
//			distributeMarketDepth(marketDepth);
		}

		@Override
		public void updateMktDepthL2(int reqId, int position, String marketMaker, int operation, int side, double price, int size) {
			if(this.id!=reqId){
				throw new IllegalArgumentException("wrong request id:"+reqId+". Was expecting reqId="+this.id);
			}
			marketDepthEventHolder.setMarketMaker(marketMaker);
			marketDepthEventHolder.setOperation(operation);
			marketDepthEventHolder.setPosition(position);
			marketDepthEventHolder.setPrice(price);
			marketDepthEventHolder.setSide(side);
			marketDepthEventHolder.setSize(size);
			distributeMarketDepthEvent(marketDepthEventHolder);
//			marketDepth.updateMktDepth(position, marketMaker, operation, side, price, size);
//			distributeMarketDepth(marketDepth);
		}
		
		@Override
		public void error( int reqId, int errorCode, String message) {
			log.info("[error]: " + reqId + " / " + errorCode + " / " + message);
			if (errorCode ==MarketDepthEvent.MKT_DEPTH_DATA_RESET){
				marketDepthEventHolder.setOperation(MarketDepthEvent.MKT_DEPTH_DATA_RESET);
				distributeMarketDepthEvent(marketDepthEventHolder);
//				marketDepth.reset();
//				distributeMarketDepth(marketDepth);
			}
		}
	}
	/**
	 * private final Map&lt;InstrumentSpecification,MarketDepthWrapper&gt; marketDepthWrappers = new HashMap&lt;InstrumentSpecification,MarketDepthWrapper&gt;();
	 */
	private final Map<InstrumentSpecification,MarketDepthWrapper> marketDepthWrappers = new HashMap<InstrumentSpecification,MarketDepthWrapper>();
	
	public void subscribeToMarketDepthStream(final IEventListener<MarketDepthEvent> listener, final InstrumentSpecification spec){
		log.info("subscribe to MarketDepth stream for: " + spec.getSymbol().toString()+" "+spec.getSecurityType().toString()+"@"+spec.getExchange() );
	
		synchronized(marketScanWrappers) {
			MarketDepthWrapper w = marketDepthWrappers.get(spec);
			if(w == null) {
				int requestId = generateRequestId();
				
				w = new MarketDepthWrapper(requestId, spec);
				marketDepthWrappers.put(spec, w);
				interceptors.put(requestId, w);
				
				Contract contract = convertToContract(spec);
				contract.m_exchange="ISLAND";//only this works at this time
				connect();				
				ecs.reqMktDepth(requestId, contract, 20);
				//log.info("reqested MarketDepth data, request id=" + w.id);
			}

			w.marketDepthEvent.addEventListener(listener);
		}
	}
	
	public void unsubscribeFromMarketDepthStream(IEventListener<MarketDepthEvent> listener, InstrumentSpecification spec) {
		log.info("unsubscribe from MarketDepth stream: " + spec.getSymbol().toString()+" "+spec.getSecurityType().toString()+"@"+spec.getExchange());
		synchronized(marketDepthWrappers) {
			MarketDepthWrapper w = marketDepthWrappers.get(spec);
			if(w == null) {
				log.warn("not subscribed");
				return;
			}
			w.marketDepthEvent.removeEventListener(listener);
			if(w.marketDepthEvent.isEmpty()) {
				ecs.cancelMktDepth(w.id);
				interceptors.remove(w.id);
				//log.info("canceled MarketDepth data for request id: " + w.id);
				// NOTE:added to make sure we get new request id's the next time we request market data!
				marketDepthWrappers.remove(spec);
			}
		}
	}
	/**
	 * returns the associated previouslyOpenedOrdersLoaded(boolean)
	 * @return
	 */
	public boolean arePreviouslyOpenedOrdersAlreadyLoaded() {
		return previouslyOpenedOrdersLoaded;
	}
	/**
	 * sets the associated previouslyOpenedOrdersLoaded(boolean) with the given previouslyOpenedOrdersLoaded(boolean)
	 * @param previouslyOpenedOrdersLoaded
	 */
	public void setPreviouslyOpenedOrdersAlreadyLoaded(boolean previouslyOpenedOrdersLoaded) {
		this.previouslyOpenedOrdersLoaded = previouslyOpenedOrdersLoaded;
	}
}
